AIエージェント基盤を止めずに更新する — Python ホットリロードと、Web サーバ層の分離

2026年9月26日

私の環境では、自分で書いたサーバ本体の上で、役割ごとのエージェントのプログラムが動いている。サーバもエージェントも、ほぼ毎日書き換わる。一日に100件を超える変更が入ることも珍しくない。

書き換えたコードをどう反映するか。素朴にやるなら、プロセスを再起動する。だが再起動のたびに進行中の会話が切れる。私の作業は会話の中で進んでいるので、これは痛い。

使っている最中に裏で中身が入れ替わり、しかも会話が途切れない。そういう更新の仕方を目指した。

たどり着いた答えは二つある。一つは、Python のコードとしてホットリロードできる範囲をできるだけ広げること。もう一つは、ホットリロードできない残る一枚（Web サーバの起動部分）を、止めずに入れ替えられる仕掛けで支えることだ。

この二つは別々の話ではない。「どこまでがコードとして入れ替えられるか」を確かめ、入れ替えられない最後の一枚だけを分離して扱う、という一連の設計である。

止めないことにこだわる理由

サーバを止めると、二種類の被害が出る。一つは、新しい接続を受け付けられない瞬間ができること。利用者がそこへアクセスすると、接続を拒否される。もう一つは、処理中のリクエストが切れること。会話の途中で裏側が入れ替われば、返答は途中で消える。

どちらも、更新の頻度が上がるほど無視できなくなる。変更が日常的に入る環境では、更新のたびに数秒止まる設計は成り立たない。止まる前提そのものを外す必要があった。

もう一つ、見落としやすい被害がある。再起動のタイミングを人手で選ぶと、その判断そのものが作業を止める。いつ止めれば安全かを見極めようとして、更新を後回しにする。後回しが積み上がると、今度は「反映されていない」状態が常態になる。止めないことは、速さの話であると同時に、正確さの話でもある。

なぜ「再起動を速くする」では足りないのか

最初に考えたのは、再起動そのものを速くする方向だった。実際、プロセスの立ち上がりは数秒で済む。だが、この方向には限界がある。

第一に、速くしても「ゼロ」にはならない。数秒の空白は残る。第二に、いちばん痛いのは立ち上がりの時間ではなく、処理中のリクエストが切れることだ。これは速さでは解決しない。第三に、頻度が上がれば、数秒の累積が無視できなくなる。一日に何度も起きる前提では、毎回数秒止まる設計は維持できない。

つまり、必要なのは「短時間で再起動する」ことではなく、「止まらない仕組みに変える」ことだった。そこで発想を変え、そもそも全部を再起動する前提をやめることにした。

更新を3つの層に分ける

更新は、影響の範囲で3つに分けられる。

エージェントごとのプログラム（役割ごとの処理、プロンプト、道具の定義）
サーバ本体の主要モジュール（会話の組み立て、道具の実行、道具箱の管理、状態管理）
サーバの起動そのもの（アプリの定義とルート）

このうち上の2つは、Python のプロセスを起こし直さなくても差し替えられる。これは Python が、モジュールを実行時に読み込める言語だからできる芸当だ。残るのは3つ目だけで、ここを変えたときだけ再起動が要る。

分け方の指針は単純だ。「どれくらいの頻度で変わるか」と「反映に何が必要か」で分ける。頻度が高く影響が狭いものは軽い手段へ。頻度が低く影響が広いものだけを重い手段へ。この対応関係を先に決めておくと、日々の変更で迷わなくなる。

第1層 — エージェントのプログラムはリクエストのたびに読み直す

いちばん頻繁に変わるのは、エージェントごとのプログラムだ。役割ごとの処理、プロンプト、使える道具の定義が、一日に何度も書き換わる。

これらは、リクエストのたびにファイルの更新時刻を見て読み直す。更新時刻が変わっていれば、その場で読み込み直してから実行する。プロセスには一切触れない。

実装は、更新時刻を鍵にした小さなキャッシュで足りる。

import importlib.util
import os

_module_cache = {}

def load_dynamic_module(name: str, path: str):
    """更新時刻が変わっていれば読み直す。変わっていなければキャッシュを返す。"""
    mtime = os.path.getmtime(path)
    cached = _module_cache.get(path)

    if cached is not None and cached["mtime"] == mtime:
        return cached["module"]          # 変更なし → 前回の読み込みを使い回す

    spec = importlib.util.spec_from_file_location(name, path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)      # ← ここで新しいコードが効になる

    _module_cache[path] = {"mtime": mtime, "module": module}
    return module

対象は、エージェントごとのディレクトリに置かれた実行コード、道具の定義、スキル文書、共通プロンプト、ナレッジ文書である。これらは名前が固定されていても、中身は毎日変わる。更新時刻だけを見て、変わっていれば読み直す。それだけで足りる。

進行中の会話は古い内容のまま最後まで走り、次の会話から新しい内容が効く。この方式の利点は、変更の粒度と反映の粒度が一致していることだ。あるエージェントの振る舞いだけを直したいとき、他のエージェントに影響が及ばない。全体を止める理由がない。

日常的な変更のほとんどは、この層で吸収される。エージェントの振る舞いを直したいとき、プロンプトを一行足したいとき、道具を一つ増やしたいときに、再起動のことを考える必要はない。

第2層 — 主要モジュールは自分自身を読み込み直す

次に変わるのが、サーバ本体の主要モジュールだ。会話の組み立て、道具の実行、道具箱の管理、状態管理の4つである。

これらは起動時に読み込まれるため、素朴に考えると再起動が要る。だが Python には、実行中のプログラムが自分自身のモジュールを読み込み直す仕組みがある。これを使えば、プロセスを止めずに差し替えられる。

import importlib
import sys

CORE_MODULES = [
    "tool_executor",      # 下流
    "toolbox_registry",
    "state_manager",
    "loop_engine",        # 上流
]

def core_reload(main_module):
    """主要モジュールを下流から順に読み込み直し、入口側の参照を張り替える。"""
    failed = []

    for name in CORE_MODULES:            # 依存される側（下流）から先に差し替える
        try:
            module = importlib.reload(sys.modules[name])
            setattr(main_module, name, module)   # ★入口が握る参照を新しい実体へ
        except Exception as exc:
            failed.append((name, repr(exc)))     # 黙って見逃さない

    return failed

ここには罠がある。読み込み直すだけでは足りない。起動の入口となるモジュールが、これらの名前を起動時に束縛している。読み込み直しても、入口側が握っている参照は古いクラスのままだ。したがって、読み込み直しと、入口側の参照の張り替え（setattr）を、必ず一組で行う。

もう一つ、順序にも意味がある。モジュール同士には依存の向きがあり、依存される側から先に差し替えないと、上流が古い下流を掴んだままになる。道具の実行、道具箱の管理、状態管理、ループ本体、という下流から上流への順で入れ替える。

大事なのは、これをやっても進行中の会話が切れないことだ。会話の処理はリクエストごとに新しく組み立てられる。進行中のリクエストは古いクラスを掴んだまま最後まで走り、次に来たリクエストから新しいコードを使う。処理中の会話を道連れにしない。

実際に検証したときは、プロセスの番号も起動時刻も変えないまま、内部のビルド識別子が切り替わることを確認した。再起動はゼロである。

失敗したときの扱いも決めてある。一部のモジュールで読み込み直しに失敗した場合、それを黙って見逃さない。失敗した対象を必ず列挙し、旧クラスが生き続ける状態に留める。部分的な適用を、成功のように扱わない。

第3層 — Web サーバ層の分離（起動部分だけが再起動対象）

ここからが「Web サーバ部分をどう分離するか」の話である。Python のコードとしてホットリロードできるのは、上の2層まで。サーバの起動そのもの、つまりアプリの定義とルートの配線を変えたときは、プロセスを立ち上げ直さないと反映されない。めったに変わらないが、ゼロではない。

この一枚には、二つの痛みが残る。接続が切れることと、処理中のリクエストが切れることだ。これを別々の仕掛けで潰した。

接続を切らさない — socket activation

一つ目には、systemd の socket activation を使った。考え方は単純で、Listen するソケットをサーバのプロセスではなく systemd に持たせる。サーバはそのソケットを引き継いで動く。

# /etc/systemd/system/ai-api.socket
[Unit]
Description=AI API listen socket

[Socket]
ListenStream=127.0.0.1:8080
Accept=no

# /etc/systemd/system/ai-api.service
[Unit]
Description=AI API server
Requires=ai-api.socket
After=ai-api.socket

[Service]
ExecStart=/opt/ai_api/.venv/bin/uvicorn main:app --fd 3
NonBlocking=true

再起動のあいだもソケットは systemd が持ち続ける。新しく来た接続は待ち行列に並び、サーバが立ち上がった時点で処理される。接続が拒否される瞬間がなくなる。

--fd 3 は、systemd が渡してくる Listen 済みソケット（既定でファイル記述子 3）を、uvicorn がそのまま使うという指定だ。サーバ側の実装には Unix ドメインソケットを前提にした記述があるが、Linux では渡された記述子がカーネルのソケットを指すため、実害はない。忠実に再現して、HTTP 200 が返るところまで確認した。

採用にあたっては、有効化の手順と、元に戻す手順を必ず対で用意した。新しい仕掛けは、止められることも同じくらい大事である。

処理中を切らない — アイドルを待つ

二つ目の「処理中のリクエストを切らない」は、再起動のタイミングをずらすことで解く。

再起動をすぐには実行しない。依頼として受け付けておき、完全に手が空いた瞬間を機械的に検知してから実行する。依頼した本人は待たされない。依頼は即座に戻り、実際の切り替えは裏で走る別のワーカーが担当する。

このワーカーは、サービス本体の管理単位から切り離しておく。だから本体を入れ替えても道連れにならず、切り替え後の確認まで一人で完走できる。

# 依頼は即座に戻す。実処理は systemd の管理単位として切り離して走らせる。
systemd-run --unit=ai-api-deferred-restart --collect \
  /opt/ai_api/scripts/deferred_restart.sh

--collect を付けておくと、ジョブが終わったあとに単位が残らない。同名の単位が既にあれば起動に失敗するので、少なくとも同時実行は防げる。

ここで重要なのは、依頼が「予約」ではなく「フラグ」として残ることだ。作業が落ち着くまで何度でも見送り、条件が揃った回にだけ実行する。判定は4つの条件を組み合わせる。

会話UIから対象ポートへの接続が残っていないこと
会話UIのログに、直近一定時間、処理要求が無いこと
会話の保存先に、直近一定時間、書き込みが無いこと
対象サービスのログが、直近一定時間、静かなこと

一つでも満たさなければ、その回は見送り、次回へ持ち越す。判定できない場合は、安全側に倒して再起動しない。

ここで一つ、方針を変えた。当初は一定時間で待ちを打ち切る設計だった。だがそれだと、実作業の最中でも再起動に踏み切ってしまう。動いている接続こそ、まさに作業中の流れだ。これを無視してはいけない。待ちは無制限にし、依頼が有効である限り持ち越す形にした。毎分、状態を確認し、空いていれば実行、そうでなければ次回に回す。依頼を取り消せば、次の確認で再起動せずに終わる。

私の環境では常時6本ほどの接続が動いている。だから「接続ゼロ」は、本当に手が空いたときしか訪れない。それを待つ設計にした。

ロールバックと検証

socket activation については、書き換える前の定義を退避しておき、問題があれば元に戻せるようにした。

検証は、開発環境で先に流し、問題がないことを確かめてから本番へ回した。適用の順序も固定した。開発に当てて動作を確認し、そのあとに本番へ当てる。逆順は禁止である。

再起動後は、サービスが立ち上がっているだけで満足せず、ポートが実際に待ち受けを開始するまで待って確認する。立ち上がった直後は、サービスは動いていてもソケットが未開通のことがある。それを待たずに成功と見なすと、「上がっていないのに成功扱い」を招く。

# 「起動済み」ではなく「外から届く」で成功を判定する
for i in $(seq 1 30); do
  if curl -fsS http://127.0.0.1:8080/health >/dev/null 2>&1; then
    echo "port is ready"; exit 0
  fi
  sleep 1
done
echo "not ready"; exit 1

反映が本当に効いたかどうかも、目視や推測ではなく、識別子の変化とプロセス情報で確かめる。起動時刻が変わっていなければ再起動していない。識別子が変わっていれば、ディスク上の新しいコードが有効になっている。この二つを毎回記録に残す。