24時間動き続けるAIエージェント：途中で「死なない」ようにする設計

なぜこれが難しいのか

AIエージェントをデモとして動かすだけなら簡単です。LLMを呼び、結果を受け取り、画面に出力する。それだけで済みます。しかし、エージェントを何時間も、何日も連続で動かし、ツール呼び出し・ファイル読み込み・外部API呼び出し・非同期処理の待機といった複雑な処理を自動で実行させようとすると、話はまったく変わってきます。

エージェントが「死ぬ」理由は無数にあります。API呼び出しのタイムアウト、キャッチされていない例外、コンテナのOOM Kill、処理途中でのネットワーク切断、あるいは単純にLLMが想定外のフォーマットを返してパース処理がクラッシュする、など。

つまり問題は「エージェントはクラッシュするのか？」ではありません。必ずクラッシュします。 本当の問題は、クラッシュしたときにシステムがそれを検知できるか、そして中断した地点から再開できるか、という点です。

信頼できるエージェントランタイムを支える3本柱

1. ハートビート / ヘルスチェック — エージェントの生死を知る

エージェント群（fleet）が今も動いているかを知る最も安価で効果的な方法は、各エージェントが一定周期で「生きている」というタイムスタンプを書き込み、それを別のウォッチャーが確認する仕組みです。

import time
import json
from pathlib import Path

HEARTBEAT_FILE = Path("agent_state/heartbeat.json")
HEARTBEAT_INTERVAL = 15  # 秒

def write_heartbeat(agent_id: str, status: str = "running"):
    HEARTBEAT_FILE.parent.mkdir(parents=True, exist_ok=True)
    HEARTBEAT_FILE.write_text(json.dumps({
        "agent_id": agent_id,
        "status": status,
        "ts": time.time(),
    }))

def agent_loop(agent_id: str):
    while True:
        write_heartbeat(agent_id)
        try:
            step_result = run_one_agent_step()
            handle_result(step_result)
        except Exception as e:
            write_heartbeat(agent_id, status=f"error: {e}")
            raise
        time.sleep(1)

ウォッチャーは独立したプロセス（別プロセス、あるいはcronジョブ）として動かし、30秒おきにこのファイルを読むだけです。

def check_agent_alive(max_staleness=60) -> bool:
    if not HEARTBEAT_FILE.exists():
        return False
    data = json.loads(HEARTBEAT_FILE.read_text())
    return (time.time() - data["ts"]) < max_staleness

実際のプロダクション環境では、ファイルの代わりにTTL付きのRedisキー（エージェントがループごとに SETEX する）を使ったり、Prometheus/Grafanaにメトリクスを送ってダッシュボードで可視化したりするのが一般的です。ただし原理はまったく同じで、「生きている」＝「つい先ほど報告があった」ということです。

2. チェックポイント — 死んでも進捗を失わない

ここが最も重要で、かつ最も見落とされやすい部分です。エージェントが状態をすべてメモリ上（通常のPython変数）に保持している場合、プロセスが落ちた瞬間に進捗はすべて消え、最初からやり直すことになります。

解決策は**状態の外部化（externalize state）**です。重要なステップごとに、状態を永続的な場所（SQLite、Redis、JSONファイル、あるいはPostgresのテーブル）へ書き出します。

import sqlite3
import json

def save_checkpoint(agent_id: str, step: int, state: dict):
    conn = sqlite3.connect("agent_checkpoints.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS checkpoints (
            agent_id TEXT PRIMARY KEY,
            step INTEGER,
            state TEXT,
            updated_at REAL
        )
    """)
    conn.execute("""
        INSERT INTO checkpoints (agent_id, step, state, updated_at)
        VALUES (?, ?, ?, ?)
        ON CONFLICT(agent_id) DO UPDATE SET
            step=excluded.step, state=excluded.state, updated_at=excluded.updated_at
    """, (agent_id, step, json.dumps(state), time.time()))
    conn.commit()
    conn.close()

def load_checkpoint(agent_id: str):
    conn = sqlite3.connect("agent_checkpoints.db")
    row = conn.execute(
        "SELECT step, state FROM checkpoints WHERE agent_id = ?", (agent_id,)
    ).fetchone()
    conn.close()
    if row is None:
        return None
    step, state_json = row
    return step, json.loads(state_json)

エージェント起動時に最初に行うべきなのは「新しいタスクを始めること」ではなく、「中断されたチェックポイントがないか確認し、あればそこから再開すること」です。

def start_or_resume(agent_id: str):
    checkpoint = load_checkpoint(agent_id)
    if checkpoint:
        step, state = checkpoint
        print(f"Resuming from step {step}")
    else:
        step, state = 0, initial_state()

    while not is_done(state):
        state = run_step(state)
        step += 1
        save_checkpoint(agent_id, step, state)

LangGraphを使っている場合、これに相当する仕組みが**永続化（persistence / checkpointer）**として標準で用意されています（MemorySaver、SqliteSaver、PostgresSaver）。各ノードの実行が終わるたびにgraph stateが自動保存され、中断されたスレッドは thread_id を指定するだけで再開できるため、上記のロジックを自前で書く必要はありません。

3. ウォッチドッグと自動再起動 — 「クラッシュ後にツールが2回実行される」問題

ハートビートとチェックポイントが揃ったら、最後は外部の監視プロセスです。エージェントの死を検知したら自動的に再起動します。

import subprocess
import time

def watchdog(agent_id: str, cmd: list[str], max_staleness=60):
    while True:
        if not check_agent_alive(max_staleness):
            print(f"[watchdog] Agent {agent_id} が応答しません。再起動します...")
            subprocess.Popen(cmd)
        time.sleep(15)

実際の環境では、これはKubernetesの livenessProbe + restartPolicy: Always、systemdの Restart=on-failure、あるいはsupervisor/PM2がすでにやってくれていることです。ウォッチドッグを自前で書く必要は必ずしもありません。しかし、ハートビートとチェックポイントは自分で実装しなければなりません。それはあなたのエージェントにしか分からない情報だからです。

ここで一つ重要な落とし穴があります。副作用を持つツール呼び出し（メール送信、トランザクション作成、決済API呼び出しなど）が冪等でない場合、エージェントが再起動して再開したときにそのステップが2回実行されてしまう可能性があります。ツールの副作用はすでに実行済みなのに、チェックポイントの保存が間に合わないままクラッシュするケースです。したがって、副作用を持つツールには冪等性キー（例：入力とステップIDのハッシュ）を持たせ、重複実行を防ぐべきです。

コンテキストプルーニング — エージェントを「肥大化」させない

エージェントが長く動くほど、会話履歴（コンテキスト）は長くなります。これは3つの問題を引き起こします。トークンコストの増加、レイテンシの増加、そしてより深刻なのは、エージェントの挙動が予測しにくくなることです。LLMが、もはや関連性のない情報を大量に含む巨大なコンテキストを処理しなければならなくなるためです。

一般的な対処法は階層的なプルーニングです。

def prune_context(messages: list[dict], keep_last_n=10, summarize_fn=None) -> list[dict]:
    system_msgs = [m for m in messages if m["role"] == "system"]
    other_msgs = [m for m in messages if m["role"] != "system"]

    if len(other_msgs) <= keep_last_n:
        return messages

    to_summarize = other_msgs[:-keep_last_n]
    recent = other_msgs[-keep_last_n:]

    summary = summarize_fn(to_summarize) if summarize_fn else "[過去のステップは省略されました]"
    return system_msgs + [{"role": "system", "content": f"これまでの進捗の要約: {summary}"}] + recent

残すのは、システムプロンプト、直近の重要なツール実行結果、そしてそれ以前に起きたことの短い要約です。履歴全体をそのまま保持するのではありません。これによってエージェントはより決定的（deterministic）に動作するようになります。同じ要約状態からは常に似た挙動が導かれるようになり、際限なく長くなりノイズが増えていくコンテキストに左右されなくなるからです。

長期稼働エージェントを作るときのチェックリスト

エージェントを無人環境に投入する前に、次の点を自問してみてください。

• エージェントは一定周期で「生きている」と報告しているか？
• 途中で死んだとき、中断地点から再開できるか？それとも最初からやり直しか？
• 副作用を持つツールはすべて冪等か？
• コンテキストは無限に肥大化しないよう剪定されているか？
• ログやメトリクスを読んで、エージェントが「動いている」だけでなく「実際に何をしているか」を把握できる人／仕組みはあるか？

---

デモとして動くエージェントと、プロダクションで動き続けるエージェントの差は、モデルの賢さではなく、こうした運用面の設計にあります。ハートビート、チェックポイント、冪等性、コンテキスト管理――この4つを最初から組み込んでおくことが、夜中にエージェントが静かに死んでいたという事態を防ぐ最短の道です。