AI Engineering Academy · 강의

체크포인트 저장과 작업 재개

각 단계를 완료할 때마다 에이전트 상태를 저장하여, 오류가 발생해도 처음부터 다시 시작하지 않고 마지막으로 성공한 체크포인트부터 장시간 실행 작업을 재개할 수 있게 합니다.

체크포인트 저장과 작업 재개은(는) CoddyKit의 무료 AI Engineering Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Engineering Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

The Problem with Long-Running Agents

An agent executing a 50-step research task might run for 30 minutes. If it fails at step 47 due to an API timeout or server restart, restarting from scratch wastes all the work done and costs tokens. Checkpointing persists the agent's state at each completed step so the task can resume from the last successful point rather than the beginning. This is essential for any agent task longer than a few minutes.

What Agent State to Persist

An agent's state consists of: the task definition, the completed steps with their tool calls and observations, the current step index, any accumulated results (files written, data collected), and metadata like start time and total token usage. Persist all of this after each step completes. State must be serializable — prefer JSON over Python objects for portability.

from dataclasses import dataclass, field
from typing import List, Any, Optional

@dataclass
class AgentStep:
    step_index: int
    thought: str
    tool_name: str
    tool_args: dict
    observation: str
    tokens_used: int
    completed_at: str

@dataclass
class AgentCheckpoint:
    task_id: str
    task_description: str
    status: str  # 'running', 'completed', 'failed'
    current_step: int
    completed_steps: List[AgentStep] = field(default_factory=list)
    accumulated_results: dict = field(default_factory=dict)
    total_tokens: int = 0
    final_answer: Optional[str] = None

Saving Checkpoints After Each Step

After each successful step, serialize the checkpoint and write it to a persistent store. Use a database or object storage (S3, Redis) rather than local disk so the checkpoint survives server restarts. Include the task_id in the key so you can retrieve a specific task's checkpoint. Also write a step log entry for every step to maintain an audit trail even if the checkpoint file gets corrupted.

import json
import redis
from dataclasses import asdict

redis_client = redis.Redis()

def save_checkpoint(checkpoint: AgentCheckpoint):
    key = f'agent:checkpoint:{checkpoint.task_id}'
    data = json.dumps(asdict(checkpoint), default=str)
    redis_client.set(key, data, ex=86400)  # 24h TTL
    # Also append to step log
    log_key = f'agent:log:{checkpoint.task_id}'
    if checkpoint.completed_steps:
        last = checkpoint.completed_steps[-1]
        redis_client.rpush(log_key, json.dumps(asdict(last), default=str))

def load_checkpoint(task_id: str) -> AgentCheckpoint | None:
    key = f'agent:checkpoint:{task_id}'
    data = redis_client.get(key)
    if data:
        return AgentCheckpoint(**json.loads(data))
    return None

Resuming from a Checkpoint

When resuming a task, load the checkpoint and reconstruct the agent's message history from the completed steps. Start the execution loop from the next uncompleted step index. The model receives its previous thoughts and observations in the message history so it has full context of what has been done, allowing it to continue without repeating work.

async def resume_or_start(task_id: str, task_description: str) -> str:
    checkpoint = load_checkpoint(task_id)
    if checkpoint and checkpoint.status == 'running':
        print(f'Resuming task {task_id} from step {checkpoint.current_step}')
        messages = rebuild_history(checkpoint)
        start_step = checkpoint.current_step
    else:
        print(f'Starting new task {task_id}')
        checkpoint = AgentCheckpoint(task_id=task_id, task_description=task_description, status='running', current_step=0)
        messages = [{'role': 'user', 'content': task_description}]
        start_step = 0
        save_checkpoint(checkpoint)
    return await run_agent_from(checkpoint, messages, start_step)

Rebuilding Message History from Steps

The key to resumption is faithfully reconstructing the message history from the persisted steps. Each completed step corresponds to an assistant message (thought + tool call) and a tool message (observation). Replay all completed steps as messages before continuing, so the model has the same context as if it had never been interrupted.

def rebuild_history(checkpoint: AgentCheckpoint) -> list:
    messages = [{'role': 'user', 'content': checkpoint.task_description}]
    for step in checkpoint.completed_steps:
        # Reconstruct the agent's reasoning message
        messages.append({
            'role': 'assistant',
            'content': f'Thought: {step.thought}\nAction: {step.tool_name}({step.tool_args})'
        })
        # Reconstruct the tool observation
        messages.append({
            'role': 'user',
            'content': f'Observation: {step.observation}'
        })
    return messages

Idempotent Tool Calls

If a step was partially completed before a crash (the tool was called but the observation was not saved), resumption might call the tool again. Design tools to be idempotent: calling them twice with the same arguments produces the same result as calling once. For write operations (creating files, sending emails), use deduplication keys to prevent duplicate effects even if the tool is called multiple times.

async def idempotent_write_file(content: str, path: str, task_id: str, step: int) -> str:
    dedup_key = f'{task_id}:step_{step}:write:{path}'
    if redis_client.exists(dedup_key):
        return f'File {path} already written (dedup key present)'
    with open(path, 'w') as f:
        f.write(content)
    redis_client.set(dedup_key, '1', ex=3600)
    return f'Successfully wrote {len(content)} chars to {path}'

Checkpoint Cleanup and Retention

Checkpoints consume storage and should not accumulate indefinitely. Set a retention policy: delete completed checkpoints after 24 hours, delete failed checkpoints after 7 days (for post-mortem analysis), and never delete running checkpoints automatically. Implement a background cleanup job that runs hourly and purges expired checkpoints according to the policy.

from datetime import datetime, timedelta

RETENTION = {
    'completed': timedelta(hours=24),
    'failed': timedelta(days=7),
    'running': None  # never auto-delete
}

def cleanup_expired_checkpoints():
    now = datetime.utcnow()
    for key in redis_client.scan_iter('agent:checkpoint:*'):
        data = json.loads(redis_client.get(key))
        status = data.get('status', 'running')
        retention = RETENTION.get(status)
        if retention is None:
            continue
        started = datetime.fromisoformat(data.get('started_at', str(now)))
        if now - started > retention:
            redis_client.delete(key)

Checkpointing in LangGraph

LangGraph has native checkpointing support via its MemorySaver and SqliteSaver classes. Attach a checkpointer to your graph and every node execution is automatically persisted. To resume, call graph.invoke with the same thread_id. LangGraph handles history reconstruction and step tracking, so you do not need to implement the checkpointing logic manually.

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph

# Create graph with persistent checkpointer
checkpointer = SqliteSaver.from_conn_string('/tmp/agent_state.db')
graph = StateGraph(AgentState)
graph.add_node('reason', reason_node)
graph.add_node('act', act_node)
# ... add edges ...
app = graph.compile(checkpointer=checkpointer)

# Run with thread_id - LangGraph auto-checkpoints
config = {'configurable': {'thread_id': 'task_abc123'}}
result = await app.ainvoke({'task': 'Research climate change'}, config)

# Resume same thread - LangGraph loads from checkpoint
result = await app.ainvoke({'task': 'Continue'}, config)

Distributed Checkpointing for Parallel Agents

When multiple agents work in parallel on sub-tasks, each agent needs its own checkpoint namespace. Use a hierarchical key structure: parent_task_id:sub_task_id. The parent agent's checkpoint records which sub-tasks have completed and their results. When the parent resumes, it re-uses completed sub-task results from the checkpoint instead of re-running them.

async def parallel_with_checkpoints(parent_id: str, subtasks: list) -> list:
    results = []
    for i, subtask in enumerate(subtasks):
        sub_id = f'{parent_id}:sub_{i}'
        # Check if subtask already completed
        existing = load_checkpoint(sub_id)
        if existing and existing.status == 'completed':
            print(f'Sub-task {i} already done, using cached result')
            results.append(existing.final_answer)
        else:
            result = await run_agent(sub_id, subtask)
            results.append(result)
    return results

Testing Resumption Behavior

Write integration tests that deliberately crash the agent mid-task and verify that resumption produces the correct final result. Simulate a crash by raising an exception at a specific step index. After resumption, check that only the steps after the crash are re-executed (not the ones before). Also test that idempotent tool calls do not produce duplicate side effects when a step is replayed.

import pytest

@pytest.mark.asyncio
async def test_resumption_from_step_3():
    task_id = 'test_resume_001'
    # Run until step 3, then crash
    with pytest.raises(SimulatedCrash):
        await run_agent_crashing_at(task_id, 'Research AI trends', crash_at_step=3)

    checkpoint = load_checkpoint(task_id)
    assert checkpoint.current_step == 3
    assert len(checkpoint.completed_steps) == 3

    # Resume and complete
    result = await resume_or_start(task_id, 'Research AI trends')
    assert result is not None
    # Verify only steps 4+ were re-executed
    assert checkpoint_step_was_not_replayed(task_id, step=0)

Checkpoint Versioning for Schema Changes

When you change the AgentCheckpoint schema (adding or renaming fields), old checkpoints in storage become incompatible. Handle this with checkpoint versioning: add a checkpoint_version field and write migration functions to upgrade old checkpoints to the new schema when they are loaded. This prevents crashes when agents resume after a deployment that changed the checkpoint format.

def load_and_migrate_checkpoint(task_id: str) -> AgentCheckpoint:
    raw = json.loads(redis_client.get(f'agent:checkpoint:{task_id}'))
    version = raw.get('checkpoint_version', '1.0')
    if version == '1.0':
        # Migrate: add new fields added in v2.0
        raw['checkpoint_version'] = '2.0'
        raw['accumulated_results'] = raw.get('accumulated_results', {})
        raw['total_tokens'] = raw.get('total_tokens', 0)
    return AgentCheckpoint(**raw)

Quick Check

Test your understanding of checkpointing and task resumption in agents.

Lesson Recap

In this lesson you learned: checkpointing serializes agent state after each step so long-running tasks survive failures without restarting from scratch, rebuilding message history from persisted steps gives the model full context on resumption, and idempotent tools with deduplication keys prevent duplicate side effects when steps are replayed. Next up we design human-in-the-loop escalation triggers.

자주 묻는 질문

“체크포인트 저장과 작업 재개” 강의는 무료인가요?

네 — “체크포인트 저장과 작업 재개” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Engineering Academy 강의 전체를 잠금 해제할 수 있습니다. AI Engineering Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

“체크포인트 저장과 작업 재개”에서 뭘 배우나요?

각 단계를 완료할 때마다 에이전트 상태를 저장하여, 오류가 발생해도 처음부터 다시 시작하지 않고 마지막으로 성공한 체크포인트부터 장시간 실행 작업을 재개할 수 있게 합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Engineering Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Engineering Academy을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Engineering Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“체크포인트 저장과 작업 재개” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Engineering Academy 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Engineering Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Engineering Academy(으)로 돌아가기