自律エージェントを24時間動かすために実装した5つの仕組み

結論

• 自律エージェントを24h動かすには、賢さより死なない仕組みが要る
• 必要なのは5つ：daemon化・フェーズ分離・interval制御・依存解決・ロールバック
• 全部Pythonで書ける。フレームワーク不要、200〜500行で組める

以下、僕がいま自分のプロジェクトで実装している5つの仕組みを、コード例つきで残しておく。完璧じゃない。けど動いてる。

1. daemon.py — 「死なないループ」を作る

自律エージェントを動かす一番外側のラッパー。1個のPythonプロセスを24時間生かし続けるためのコア。

なんでwhile Trueじゃダメかというと、例外で1回でも落ちたらそこで終わるから。systemd で再起動すればいい派もいるけど、僕はプロセス内で復帰するほうが状態を引き継げて好き。

# coo/daemon.py（最小版）
import time
import traceback
from datetime import datetime

def run_forever(executor, interval_sec=300):
    """1個のexecutorを死なせずに回し続ける"""
    while True:
        try:
            executor()  # 1サイクル分の仕事
        except KeyboardInterrupt:
            print("[daemon] stop requested")
            break
        except Exception as e:
            # 落ちても止めない。ログだけ残して次のサイクルへ
            print(f"[daemon] {datetime.now()} error: {e}")
            traceback.print_exc()
        time.sleep(interval_sec)

ポイントは KeyboardInterrupt だけは透過させること。Ctrl+C で止められないデーモンはデバッグ不能になる。自分が止められない自動化は、自動化じゃなくて事故。

2. phase: boot / continuous の分離

エージェントには「起動時に1回だけやること」と「継続的にやり続けること」がある。これを混ぜると、再起動のたびに初期化処理が走って重複が出たり、逆に継続処理が止まったりする。

僕は plan.json に phase フィールドを足して分けている。

{
  "agents": [
    {
      "id": "agent_init",
      "role": "状態リセット・キャッシュ削除",
      "phase": "boot",
      "task": "前回の途中状態をクリーンアップ",
      "output_file": "results/00_boot.md"
    },
    {
      "id": "agent_poll",
      "role": "毎時のRSS取得",
      "phase": "continuous",
      "interval_sec": 3600,
      "task": "RSSフィードから新着を収集",
      "output_file": "results/01_feed.md"
    }
  ]
}

オーケストレータ側で読み分ける。

# coo/orchestrator.py（抜粋）
def run_project(plan):
    # bootフェーズ：起動時1回だけ
    for a in plan["agents"]:
        if a.get("phase") == "boot":
            execute_agent(a)

    # continuousフェーズ：それぞれのintervalで回す
    continuous = [a for a in plan["agents"]
                  if a.get("phase") == "continuous"]
    schedule_loop(continuous)

boot を分けた瞬間、再起動の安全性が一段上がった。初期化処理を冪等に書く必要が薄れる。「ここは1回しか走らない」と言える保証は、設計を相当ラクにする。

3. interval制御 — エージェントごとにリズムを変える

agent_A は1分ごと、agent_B は1時間ごと、agent_C は1日1回。これを同じループで回したい。

雑な実装だと「最小単位（1分）で全部回す」だけど、agent_C まで1分ごとに呼ぶのは無駄だし、外部APIのレート制限を喰う。

僕の実装はこう。各エージェントに interval_sec と last_run を持たせる。

# coo/scheduler.py
import time

def schedule_loop(agents):
    state = {a["id"]: {"last_run": 0} for a in agents}
    while True:
        now = time.time()
        for a in agents:
            interval = a.get("interval_sec", 3600)
            if now - state[a["id"]]["last_run"] >= interval:
                try:
                    execute_agent(a)
                    state[a["id"]]["last_run"] = now
                except Exception as e:
                    # 失敗してもlast_runは更新しない=次サイクルで即リトライ
                    print(f"[sched] {a['id']} failed: {e}")
        time.sleep(30)  # 30秒の解像度で十分
    # ちなみに「sleep 30」はループ全体の最小粒度。
    # 1分intervalのエージェントも実際は30〜60秒の揺れで走る。許容する。

ループ自体の sleep は30秒くらいで十分。1秒単位の精度が要るならそれは自律エージェントじゃなくてリアルタイムシステムなので別の話。

4. 依存解決 — depends_on で順序を守る

複数エージェントが連携するとき、「Bは Aの出力を読む」みたいな依存が出る。

最初は雑に「順番に書いた順で実行」していた。これは1回目はいいけど、interval がバラバラになると壊れる。AがまだのときにBが走ると、Bは古い出力を読む。

depends_on を導入した。

{
  "id": "agent_summarize",
  "phase": "continuous",
  "interval_sec": 3600,
  "depends_on": ["agent_poll"],
  "task": "agent_pollの結果を要約",
  "output_file": "results/02_summary.md"
}

実装側はトポロジカルソートで実行順を決める。

# coo/depgraph.py
from collections import defaultdict, deque

def topo_sort(agents):
    deps = {a["id"]: a.get("depends_on", []) for a in agents}
    in_deg = defaultdict(int)
    graph = defaultdict(list)
    for aid, ds in deps.items():
        for d in ds:
            graph[d].append(aid)
            in_deg[aid] += 1

    q = deque([a["id"] for a in agents if in_deg[a["id"]] == 0])
    order = []
    while q:
        x = q.popleft()
        order.append(x)
        for y in graph[x]:
            in_deg[y] -= 1
            if in_deg[y] == 0:
                q.append(y)
    if len(order) != len(agents):
        raise RuntimeError("依存に循環がある")
    return order

地味だけど、これ入れた瞬間に順序起因のバグが消えた。デバッグ時間が一気に短くなる。

ちなみに循環依存は実行時じゃなくて起動時にエラーにする。動いてから「ぐるぐる回ってる」と気づくのは最悪のパターン。

5. 失敗時ロールバック — 「出力ファイルを途中で壊さない」

エージェントが書き込み途中で死ぬと、output_file が半分書かれた壊れた状態で残る。次の依存先がこれを読むと連鎖事故。

対策はアトミック書き込み。一時ファイルに書いて、最後にrenameする。

# coo/io_safe.py
import os
import tempfile

def write_atomic(path: str, content: str):
    """書き込みが完了したファイルだけがpathに現れる"""
    dirname = os.path.dirname(path) or "."
    os.makedirs(dirname, exist_ok=True)
    # 同じディレクトリにtmp（rename はファイルシステム跨ぐと非アトミック）
    fd, tmp = tempfile.mkstemp(dir=dirname, prefix=".tmp_", suffix=".part")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())  # OSバッファまで書き込み待つ
        os.replace(tmp, path)  # POSIX/Windows両方でアトミック
    except Exception:
        try:
            os.remove(tmp)
        except OSError:
            pass
        raise

os.replace は Windows でもアトミック。os.rename は Windows で上書き不可なので注意（これで一度ハマった。Linux ではテスト通って、Windows でだけ壊れる地獄）。

ロールバック観点で言うと、ここに加えて「1世代前の出力を残す」ようにもしている。output.md を書き換える前に output.prev.md にコピーする。事故ったら手動で戻せる。

def write_versioned(path: str, content: str):
    if os.path.exists(path):
        backup = path.replace(".md", ".prev.md")
        os.replace(path, backup)
    write_atomic(path, content)

失敗体験：止め方を実装し忘れて3日動き続けた

笑い話だけど、最初に作ったときは止め方を実装し忘れた。KeyboardInterrupt をうっかり try/except で握りつぶしていて、Ctrl+C が効かない。

WSL のターミナルを閉じても、nohup 相当の挙動で生き続けて、3日後に「あれ、API のクレジット結構減ってる」で気づいた。ps aux | grep daemon でPID 探して kill -9 で止めた。

教訓：止め方を最初に実装する。動かす前に。

まとめ

5つ並べた：

1. daemon.py — 例外で死なないループ
2. boot / continuous — 起動時1回と継続を分ける
3. interval制御 — エージェントごとのリズム
4. depends_on — 順序保証、循環は起動時拒否
5. アトミック書き込み + 世代管理 — 壊れた出力を残さない

これ以上のことは正直あまり要らない。コードを増やすほど運用が重くなる。「賢いエージェント」より「死なないエージェント」が、月単位で見ると圧倒的に勝つ。

複雑なフレームワーク入れる前に、この5つを200行くらいで自分で書くのを勧めたい。書いた経験が、後でフレームワーク選ぶときの判断軸になる。

---

次回予告

次は「自律エージェントを止めずにアップデートする」やり方を書く。daemon を動かしたままコードを差し替える方法、SIGHUP で設定だけリロードする方法、plan.json のホットリロード。止めずに進化させるのが次の課題で、いま実装中。

— Sai

---

この記事が役に立ったら：僕が自律エージェントを動かすときに実際に使っているプロンプトを2つのパックにまとめました — 自律エージェント用プロンプト100選 と Claude Code パワーユーザー向けプロンプト。どれも「まず動かす」発想で、ターミナルに貼って即使えます。