워크플로 체크포인트

Microsoft Agent Framework 워크플로 - 체크포인트

이 페이지는 Microsoft Agent Framework 워크플로 시스템의 **체크포인트(Checkpoints)**에 대한 개요를 제공합니다.

개요

체크포인트는 워크플로 실행 중 특정 시점에 워크플로 상태를 저장하고, 나중에 그 지점부터 이어서 실행할 수 있게 해줍니다. 이 기능은 특히 다음 시나리오에서 유용해요.

  • 실패가 나도 진행 상황을 잃고 싶지 않은 장기 실행 워크플로.
  • 실행을 잠시 멈췄다가 나중에 이어가고 싶은 장기 실행 워크플로.
  • 감사나 규정 준수 목적으로 주기적으로 상태를 저장해야 하는 워크플로.
  • 서로 다른 환경이나 인스턴스로 마이그레이션해야 하는 워크플로.

체크포인트는 언제 만들어질까?

워크플로는 워크플로 실행 모델에 문서화된 대로 슈퍼스텝(superstep) 단위로 실행된다는 점을 기억하세요. 체크포인트는 각 슈퍼스텝이 끝날 때, 그 슈퍼스텝의 모든 실행기가 실행을 마친 뒤에 만들어집니다. 체크포인트는 워크플로의 전체 상태를 담는데, 여기에는 다음이 포함됩니다.

  • 모든 실행기의 현재 상태
  • 다음 슈퍼스텝을 위한 워크플로의 모든 대기 중 메시지
  • 대기 중인 요청과 응답
  • 공유 상태

::: zone pivot="programming-language-python"

[!NOTE] Python 버전 1.13.0부터 워크플로는 첫 슈퍼스텝 전에 워크플로 입력을 기록하는 엔트리 체크포인트를, 그리고 요청 이벤트에 대한 응답이 전달될 때 또 하나의 엔트리 체크포인트를 추가로 만듭니다. 이 체크포인트들 덕분에 전체 워크플로 실행을 리플레이할 수 있어요. 이 릴리스에는 반복 횟수, 메시지 소스 ID, 체크포인트 순서에 의존하는 애플리케이션에 대한 사소한 하위 호환성 변경이 포함됩니다. 기존 체크포인트는 계속 지원돼요. 마이그레이션 방법은 Python 워크플로 체크포인트를 1.13.0으로 업그레이드하기를 참고하세요.

::: zone-end

체크포인트 캡처하기

::: zone pivot="programming-language-csharp"

체크포인트를 활성화하려면 워크플로를 실행할 때 CheckpointManager를 제공해야 합니다. 그러면 SuperStepCompletedEvent를 통하거나, run의 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;

::: zone-end

::: zone pivot="programming-language-python"

체크포인트를 활성화하려면 워크플로를 만들 때 CheckpointStorage를 제공해야 합니다. 그러면 스토리지를 통해 체크포인트에 접근할 수 있어요. Agent Framework는 내장 구현 세 가지를 제공합니다 — 지속성과 배포 요구에 맞는 것을 고르면 됩니다.

공급자 패키지 지속성 적합한 경우
InMemoryCheckpointStorage agent-framework 프로세스 내부 전용 테스트, 데모, 수명이 짧은 워크플로
FileCheckpointStorage agent-framework 로컬 디스크 단일 머신 워크플로, 로컬 개발
CosmosCheckpointStorage agent-framework-azure-cosmos Azure Cosmos DB 프로덕션, 분산, 프로세스 간 워크플로

셋 모두 같은 CheckpointStorage 프로토콜을 구현하므로, 워크플로나 실행기 코드를 바꾸지 않고도 공급자를 갈아끼울 수 있어요.

In-Memory

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)

allowed_checkpoint_types 파라미터로 역직렬화할 수 있는 Python 유형을 제한하는 방법은 보안 고려 사항 섹션을 참고하세요.

Azure Cosmos DB

CosmosCheckpointStorage는 체크포인트를 Azure Cosmos DB NoSQL에 영속화합니다. 지속적이고 프로세스 간 체크포인팅이 필요한 프로덕션 및 분산 워크플로에 가장 잘 맞아요. 옵션 공급자 패키지를 설치합니다.

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

데이터베이스와 컨테이너는 처음 사용할 때 자동으로 생성되고, /workflow_name이 파티션 키로 사용되어 워크플로별 쿼리를 효율적으로 만듭니다. 권장 인증 방식은 DefaultAzureCredential 같은 Azure TokenCredential을 통한 관리 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",
)

연결 정보는 전적으로 환경 변수로도 제공할 수 있습니다.

변수 설명
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=)도 받아들입니다.

::: zone-end

::: zone pivot="programming-language-go"

체크포인트를 활성화하려면 체크포인트 관리자로 실행 환경을 구성합니다. 그러면 체크포인트를 workflow.SuperStepCompletedEvent에서, 또는 run의 체크포인트 목록을 통해 접근할 수 있어요.

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

::: zone-end

체크포인트에서 이어서 실행하기

::: zone pivot="programming-language-csharp"

같은 run에서 특정 체크포인트부터 워크플로를 이어서 실행할 수 있어요.

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

::: zone-end

::: zone pivot="programming-language-python"

같은 워크플로 인스턴스에서 특정 체크포인트부터 워크플로를 이어서 실행할 수 있어요.

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

::: zone-end

::: zone pivot="programming-language-go"

같은 run에서 특정 체크포인트로 스트리밍 run을 복원할 수 있어요.

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

::: zone-end

체크포인트로 리하이드레이션하기

리하이드레이션된 워크플로는 체크포인트를 만든 워크플로의 토폴로지와 실행기 ID를 보존해야 합니다. 실행기 ID가 어떻게 해석되는지는 SDK와 실행기 유형에 따라 달라져요.

::: zone pivot="programming-language-csharp"

또는 체크포인트에서 새 run 인스턴스로 워크플로를 리하이드레이션할 수 있습니다.

:::code language="csharp" source="~/../agent-framework-code/dotnet/samples/03-workflows/Checkpoint/CheckpointAndRehydrate/Program.cs" id="rehydrate_workflow":::

[!IMPORTANT] ResumeStreamingAsync에 넘기는 워크플로는 체크포인트를 만든 워크플로와 동일한 구조와 실행기 ID를 가져야 합니다. 워크플로가 요청·의존성 주입 범위·프로세스·배포를 가로질러 재구성되는 로컬 ChatClientAgent 인스턴스를 담고 있다면, 각 에이전트에 안정적인 ChatClientAgentOptions.Id를 할당하세요. 에이전트가 Name도 설정한다면 그 Name도 그대로 유지해야 합니다.

예를 들어 에이전트의 논리적 역할을 나타내는 ID를 할당할 수 있습니다.

:::code language="csharp" source="~/../agent-framework-code/dotnet/samples/03-workflows/Orchestration/Handoff/AgentRegistry.cs" id="stable_agent_identity":::

이 패턴을 워크플로에 참여하는 모든 에이전트에 적용하세요. 에이전트 ID는 워크플로 안에서 고유해야 하고, 같은 논리적 에이전트를 재구성할 때 재사용되어야 합니다. 에이전트 ID로 대화 ID·요청 ID·사용자 ID·개인 식별 정보·비밀을 쓰지 마세요.

에이전트 Name이 설정된 경우, 현재 .NET 워크플로 실행기 ID는 Name과 Id 둘 다에서 파생됩니다. 따라서 둘 중 하나를 바꾸면 재빌드된 워크플로가 체크포인트와 호환되지 않게 돼요. 안정적인 값을 할당한다고 해서 다른 값이나 무작위로 생성된 ID로 만들어진 체크포인트가 복구되지는 않습니다. 그러한 경우에는 새 세션과 체크포인트 계보를 시작해야 합니다.

관련 시나리오는 에이전트로서의 워크플로와 핸드오프 오케스트레이션을 참고하세요.

::: zone-end

::: zone pivot="programming-language-python"

또는 체크포인트에서 새 워크플로 인스턴스를 리하이드레이션할 수 있습니다.

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

::: zone-end

::: zone pivot="programming-language-go"

또는 체크포인트에서 새 워크플로 인스턴스를 리하이드레이션할 수 있습니다.

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

::: zone-end

실행기 상태 저장

::: zone pivot="programming-language-csharp"

실행기의 상태가 체크포인트에 담기게 하려면, 실행기가 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);
}

::: zone-end

::: zone pivot="programming-language-python"

실행기의 상태가 체크포인트에 담기게 하려면, 실행기가 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", [])

::: zone-end

::: zone pivot="programming-language-go"

실행기 상태가 체크포인트에 담기게 하려면 실행기에 체크포인트 훅을 붙이고 워크플로 컨텍스트를 통해 상태를 저장합니다.

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

::: zone-end

보안 고려 사항

[!IMPORTANT] 체크포인트 스토리지는 신뢰 경계(trust boundary)입니다. 내장 스토리지 구현을 쓰든 커스텀 구현을 쓰든, 스토리지 백엔드는 신뢰할 수 있는 전용 인프라로 취급해야 합니다. 신뢰할 수 없거나 변조 가능성이 있는 소스에서 체크포인트를 로드하지 마세요.

::: zone pivot="programming-language-csharp"

체크포인트에 사용하는 스토리지 위치가 적절히 보호되도록 하세요. 승인된 서비스와 사용자만 체크포인트 데이터에 대한 읽기·쓰기 접근 권한을 가져야 합니다.

::: zone-end

::: zone pivot="programming-language-python"

Pickle 직렬화

FileCheckpointStorage와 CosmosCheckpointStorage는 모두 Python의 pickle 모듈을 사용해서 dataclass·datetime·커스텀 객체 같은 JSON 네이티브가 아닌 상태를 직렬화합니다. 역직렬화 중 임의 코드 실행의 위험을 줄이기 위해, 두 공급자는 기본적으로 **제한된 unpickler(restricted unpickler)**를 사용합니다. 역직렬화 중에는 안전한 내장 Python 유형(기본형, datetime, uuid, Decimal, 일반 컬렉션 등)과 지원되는 Agent Framework 또는 OpenAI SDK 유형만 허용됩니다. 모듈 접두사 허용 목록은 타입 전용입니다. 헬퍼 함수나 다른 비타입 전역 객체는 거부됩니다. 지원되지 않는 유형이 있으면 역직렬화가 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",
    ],
)

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 컨테이너에는 승인된 주체만 읽기·쓰기 접근 권한을 가져야 합니다.

::: zone-end

::: zone pivot="programming-language-go"

Go 체크포인트 관리자는 체크포인트 상태를 JSON으로 직렬화하지만, 체크포인트 스토리지는 여전히 신뢰된 애플리케이션 상태입니다. checkpoint.NewFileSystemJSONStore를 쓴다면 체크포인트 파일을 보호된 디렉터리에 저장하고 읽기·쓰기 접근을 승인된 프로세스로만 제한하세요. 커스텀 스토어는 자체적으로 접근 제어·무결성·지속성 보장을 책임집니다.

::: zone-end

다음 단계