CrewAI チェックポイント搭載途中クラッシュを無駄にしない方法

AIエージェントのワークフローが47分動き続け、312回のLLM呼び出しを経てステップ8でクラッシュした。ほとんどのフレームワークはゼロから再実行を強いる。CrewAI 1.14は、この問題に正面から答えるチェックポイント機能を搭載した。

この記事でわかること：

CrewAI チェックポイントが解決する「ゼロからの再実行」問題
CheckpointConfig の設定方法とストレージの選び方
クラッシュ後にワークフローを途中から再開する手順
フォーク機能で「あのステップからやり直す」を実現する方法
CLI の TUI でチェックポイントを管理する方法

https://docs.crewai.com/en/concepts/checkpointing

長時間ワークフローが抱える問題

マルチエージェントのワークフローは、タスクが増えるほど実行時間も伸びる。途中で API レートリミットに引っかかったり、ネットワーク障害が起きたりすると、すべての出力が無駄になる。完了済みのタスクを再実行するコストはトークン単位で積み上がり、時間だけでなく費用にも直結する。

CrewAI は 2026年4月7日リリースの v1.14.0 でチェックポイント機能を導入した。タスクが完了するたびに実行状態をファイルに保存し、失敗時はその地点から再開できる。Google ドキュメントの自動保存を AI ワークフローに持ち込んだ仕組みと考えるとわかりやすい。

チェックポイントの有効化

最短の設定は checkpoint=True を Crew に渡すだけだ。

from crewai import Crew, CheckpointConfig

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=True,  # デフォルト: ./.checkpoints/ に保存、task_completed で発火
)
result = crew.kickoff()

タスクが1つ完了するたびに ./.checkpoints/ 以下に JSON ファイルが書き出される。保存先・発火タイミング・最大保存数を細かく制御するには CheckpointConfig を使う。

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=CheckpointConfig(
        location="./my_checkpoints",
        on_events=["task_completed", "crew_kickoff_completed"],
        max_checkpoints=5,
    ),
)

max_checkpoints を指定すると古いファイルが自動で削除されるため、ディスクを圧迫しにくい。

checkpoint フィールドは Crew だけでなく Flow と Agent にも設定できる。Agent 単体で checkpoint=False を渡せばその Agent だけ除外でき、親の設定を引き継がない。

ストレージの選び方

CrewAI は 2 種類のストレージプロバイダを標準搭載している。

JsonProvider（デフォルト）はタスクごとに独立した JSON ファイルを書き出す。中身を人間が直接読めるため、デバッグや検査がしやすい。チェックポイントのファイル名は <タイムスタンプ>_<UUID>.json 形式になる。

SqliteProvider はチェックポイントをすべて 1 つの SQLite ファイルにまとめる。高頻度で発火させる場合や、ファイル数を増やしたくない場合に向いている。WAL ジャーナルモードで動作するため、書き込み中も別プロセスからの読み取りが止まらない。

from crewai import Crew, CheckpointConfig
from crewai.state import SqliteProvider

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=CheckpointConfig(
        location="./.checkpoints.db",
        provider=SqliteProvider(),
        max_checkpoints=50,
    ),
)

on_events に "*" や "llm_call_completed" を指定すると LLM 呼び出しのたびにチェックポイントが走るため、ディスクへの書き込みが非常に多くなる。その場合は必ず max_checkpoints で上限を設けること。

クラッシュ後の再開手順

再開は from_checkpoint パラメータにチェックポイントのパスを渡すだけだ。

from crewai import Crew, CheckpointConfig

crew = Crew(agents=[...], tasks=[...])

result = crew.kickoff(
    from_checkpoint=CheckpointConfig(
        restore_from="./my_checkpoints/20260407T120000_abc123.json",
    ),
)

完了済みのタスクはスキップされ、クラッシュしたタスクから再実行が始まる。restore_from 以外の CheckpointConfig フィールドは新しいランに引き継がれるため、再開後もチェックポイントの書き出しが続く。

クラスメソッドを使う書き方もある。

config = CheckpointConfig(
    restore_from="./my_checkpoints/20260407T120000_abc123.json"
)
crew = Crew.from_checkpoint(config)
result = crew.kickoff()

フォーク：別の分岐を試す

チェックポイントから「別の経路を試す」のがフォーク機能だ。

config = CheckpointConfig(
    restore_from="./my_checkpoints/20260407T120000_abc123.json"
)
crew = Crew.fork(config, branch="experiment-a")
result = crew.kickoff(inputs={"strategy": "aggressive"})

フォークした実行は固有のリネージ ID を持つため、元のチェックポイントと混在しない。branch ラベルは省略すると自動生成される。

この機能はプロンプトや入力の違いが出力にどう影響するかを比較するときに役立つ。同じステップから複数の分岐を起こし、それぞれの結果を確認できる。

CLI でチェックポイントを管理する

crewai checkpoint コマンドを実行するとターミナル上に TUI が起動する。JSON ファイルと SQLite ファイルのどちらでも自動判別して読み込む。

# デフォルト（.checkpoints/ または .checkpoints.db を自動検出）
crewai checkpoint

# 場所を明示する場合
crewai checkpoint --location ./my_checkpoints

# リスト表示のみ
crewai checkpoint list ./my_checkpoints

# 特定のチェックポイントの詳細を確認
crewai checkpoint info ./my_checkpoints/20260407T120000_abc123.json

TUI の左ペインはツリービューになっており、ブランチごとにチェックポイントがまとめられ、フォークは元のチェックポイントの下に入れ子で表示される。チェックポイントを選択すると右ペインにメタデータ・エンティティの状態・タスクの進捗が表示される。

入力値とタスク出力はその場で編集できる。タスク出力を書き換えてフォークすると、変更以降のタスクが無効化され新しい値を引き継いで再実行される。「あのタスクの出力がこうだったら後続はどうなるか」を手軽に試せる。

まとめ

CrewAI 1.14 のチェックポイント機能は、長時間ワークフローの再実行コストを削減する実用的なアップデートだ。checkpoint=True の1行で有効化でき、JSON と SQLite の2種類のストレージから用途に合わせて選べる。再開・フォーク・CLI 管理まで一通りそろっており、複数タスクを直列につなぐワークフローで特に効果が出やすい。

CrewAI の公式ドキュメントには「early release」と記載があり、API は今後変わる可能性がある。本番環境に組み込む前にバージョンの固定と動作確認を済ませておくと安全だ。