0Pricing
AI Engineering Academy · レッスン

チェックポイントとタスクの再開

完了した各ステップでエージェントの状態を永続化し、失敗時に最初からやり直すのではなく、最後に成功したチェックポイントから長時間実行タスクを再開できるようにします

「チェックポイントとタスクの再開」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応の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時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。

「チェックポイントとタスクの再開」で何を学びますか？

完了した各ステップでエージェントの状態を永続化し、失敗時に最初からやり直すのではなく、最後に成功したチェックポイントから長時間実行タスクを再開できるようにします ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Engineering Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。

「チェックポイントとタスクの再開」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Engineering Academyレッスンでコードを書いて実行できますか？

はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. エージェントの失敗モードの分類
  2. 自己修正とリフレクティブプロンプティング
  3. チェックポイントとタスクの再開
  4. Human-in-the-Loopエスカレーション
← AI Engineering Academyに戻る