AIエージェントの状態管理 — runtime_status.jsonからLangGraphまで

AIエージェントが複雑なタスクを遂行するには「今どの段階にいるか」を正確に把握する状態管理が不可欠になる。JSON単一ファイルからグラフベースの状態管理まで、規模に応じた選択肢を解説する。

状態管理の3段階

段階	方法	適用規模	複雑度
レベル1	JSONファイル	単一エージェント	低
レベル2	SQLite + JSON	複数タスク並行	中
レベル3	LangGraph	マルチエージェント	高

レベル1: runtime_status.json

最もシンプルな方法。1つのJSONファイルでエージェントの状態を管理する。

{
  "agent_id": "cloudworks-bot",
  "status": "running",
  "current_task": {
    "id": "task-001",
    "type": "scrape_and_apply",
    "phase": "applying",
    "target": "cloudworks",
    "started_at": "2026-04-09T10:00:00Z"
  },
  "last_completed": {
    "id": "task-000",
    "completed_at": "2026-04-09T09:45:00Z",
    "result": "applied_3_jobs"
  },
  "error_count": 0,
  "last_heartbeat": "2026-04-09T10:05:00Z"
}

状態遷移の定義

idle -> running -> completing -> idle
             \-> error -> recovering -> idle
                                    \-> halted（手動復旧必要）

状態遷移は必ず一方向に定義する。error -> runningのような逆戻りはrecoveringを経由させる。

読み書きのコード例

import json
from pathlib import Path
from datetime import datetime

STATUS_FILE = Path("runtime_status.json")

def update_status(phase, **kwargs):
    status = json.loads(STATUS_FILE.read_text())
    status["current_task"]["phase"] = phase
    status["last_heartbeat"] = datetime.now().isoformat()
    status.update(kwargs)
    STATUS_FILE.write_text(json.dumps(status, indent=2, ensure_ascii=False))

def get_status():
    return json.loads(STATUS_FILE.read_text())

レベル2: SQLite + JSONの併用

タスクが複数並行する場合、JSONファイルでは競合が起きる。SQLiteでタスク一覧を管理し、各タスクの詳細はJSON列に格納する。

import sqlite3, json

db = sqlite3.connect("agent_state.db")
db.execute(
    "CREATE TABLE IF NOT EXISTS tasks ("
    "id TEXT PRIMARY KEY, "
    "status TEXT DEFAULT 'pending', "
    "payload JSON, "
    "created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, "
    "updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"
)

def add_task(task_id, payload):
    db.execute("INSERT INTO tasks (id, payload) VALUES (?, ?)",
               (task_id, json.dumps(payload)))
    db.commit()

def update_task_status(task_id, status):
    db.execute("UPDATE tasks SET status=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
               (status, task_id))
    db.commit()

レベル3: LangGraphとの比較

LangGraphはグラフベースの状態管理フレームワークで、条件分岐・並行実行・人間承認を宣言的に定義できる。

比較項目	runtime_status.json	LangGraph
セットアップ	ゼロ（ファイル1つ）	pip install + グラフ定義
状態遷移	コード内にif文	グラフのエッジで宣言的定義
並行処理	ファイルロック必要	ビルトインサポート
人間承認（HITL）	自前実装	interrupt()で標準サポート
可視化	なし	グラフの可視化ツールあり
依存関係	なし	langgraph, langchain_core

選択基準

タスクが1種類で直線的 -> レベル1のJSONで十分
複数タスクが並行し、状態を永続化したい -> レベル2のSQLite
条件分岐・承認フロー・マルチエージェント連携が必要 -> レベル3のLangGraph

heartbeatパターン

エージェントが生存しているかを確認する仕組み。last_heartbeatを定期更新し、一定時間更新がなければ異常とみなす。

from datetime import datetime, timedelta

def is_agent_alive(status, threshold_minutes=10):
    last = datetime.fromisoformat(status["last_heartbeat"])
    return datetime.now() - last < timedelta(minutes=threshold_minutes)

watchdogプロセスがheartbeatを監視し、停止を検知したら再起動またはアラートを発行する。

設計のベストプラクティス

最小限の状態から始める: 最初からLangGraphを入れない。JSONで足りなくなってから移行する
状態遷移図を先に描く: コードより先に状態と遷移を定義する
heartbeatは必須: 自律エージェントは「黙って死ぬ」ことがある
エラー状態を明示的に定義する: 「不明な状態」を許容しない