DockerコンテナでAIエージェントの呼び出しコストを削減する方法
エージェントに対して subprocess や exec() でBashを直接渡すのは、家の鍵を丸ごと渡すようなものです。プロンプトインジェクション攻撃一発でホストサーバー全体が突破されたり、エージェントがハルシネーションを起こして rm -rf / を実行してしまったりする可能性があります。かといって、すべてのツールをOpenAPIスキーマとして細かく分割して登録するとどうでしょう?ツールを数個追加しただけで、LLMのコンテキストウィンドウが溢れかえってしまいます。
結局のところ、答えはセッションごとに0.1秒で起動して消えるエフェメラル(Ephemeral:使い捨て型)Dockerコンテナを接続することです。セキュリティ領域を明確に分離し、大容量 データをコンテナ内部のパイプラインで処理することで、APIトークン消費量を70%以上削減できます。
Ephemeral Bashサンドボックスの構築
エージェントが無限ループを回したり、フォークバームを爆発させたりしても、ホストサーバーは無事である必要があります。LinuxカーネルのCgroupsレベルでCPUとメモリを強制的に制限しておかないと、1つのエージェントスレッドが暴走してクラウドアカウントの破産(コスト爆発)につながります。
Python Docker SDKで構築する隔離サンドボックス構造です。
`python
import atexit
import os
import signal
import docker
from docker.errors import DockerException
class EphemeralBashSandbox:
def init(self, workspace_host_path: str, image: str = "python:3.11-slim"):
self.client = docker.from_env()
self.image = image
self.workspace_host_path = os.path.abspath(workspace_host_path)
self.container = None
self._start_sandbox()
atexit.register(self.cleanup)
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
def _start_sandbox(self):
self.container = self.client.containers.run(
image=self.image,
command="/bin/bash",
detach=True,
stdin_open=True,
tty=True,
network_disabled=True,
read_only=True,
mem_limit="512m",
cpu_quota=50000,
pids_limit=50,
user="1000:1000",
volumes={
self.workspace_host_path: {
"bind": "/workspace",
"mode": "rw"
},
"/tmp": {
"bind": "/tmp",
"mode": "rw"
}
},
working_dir="/workspace",
environment={"HOME": "/tmp"}
)
def execute_command(self, cmd: str, timeout: int = 30) -> tuple[int, str, str]:
if not self.container:
raise RuntimeError("Sandbox container is not active.")
exec_res = self.container.exec_run(
cmd=["/bin/bash", "-c", cmd],
workdir="/workspace",
demux=True
)
exit_code = exec_res.exit_code
stdout = exec_res.output[0].decode('utf-8', errors='replace') if exec_res.output and exec_res.output[0] else ""
stderr = exec_res.output[1].decode('utf-8', errors='replace') if exec_res.output and exec_res.output[1] else ""
return exit_code, stdout, stderr
def cleanup(self):
if self.container:
try:
self.container.stop(timeout=2)
self.container.remove(force=True)
except DockerException:
pass
finally:
self.container = None
def _signal_handler(self, signum, frame):
self.cleanup()
os._exit(0)
`
命令を実行するたびに docker run を新しく叩くと、4.7秒という悲惨なコールドスタートレイテンシが発生します。これではプロダクションサービスには絶対に使いものになりません。代わりにバックグラウンドデーモンモード(detach=True, stdin_open=True, tty=True)でサンドボックスを立ち上げておき、exec_run でコマンドだけを注入することで、応答速度を100ms未満に抑えることができます。
重要な設定ポイントは3つです。
mem_limit="512m"、cpu_quota=50000、pids_limit=50 でハードウェアリソースを小型スレッドレベルに制限します。
network_disabled=True と read_only=True で内部ネットワークへのアクセスを防ぎ、作業用の /workspace と /tmp 領域のみを制限付きでマウントします。
user="1000:1000" で non-root 権限を強制し、POSIXシグナルハンドラーを登録してプロセスが終了した際にコンテナが綺麗に破棄されるようにします。
OpenAPIスキーマの代わりにCLIパイプラインでトークンを節約する
個別のAPIスキーマをJSONで定義して注入する従来の方式は、ツール1つあたり550〜1,400トークンを消費します。ツールを20個追加するだけで、質問を1つ投げる前に20,000トークンが消えて飛びます。
検索エンジンYou.comの技術レポートによると、単なるJSONスキーマ注入の代わりにBashベースのスクリプト実行方式(CodeAct)を導入したところ、トークン消費量が61%削減され、処理速度は40%高速化しました。
| 評価項目 |
JSONスキーマ注入 |
Model Context Protocol (MCP) |
Bash CLIパイプライン |
| ツール定義トークン |
ツールあたり ~550–1,400 トークン |
ツールあたり ~550–1,400 トークン |
メタインターフェース1個 (~100 トークン) |
| 中間データのコンテキスト汚染 |
非常に高い (ペイロード全体を送信) |
高い (ペイロード全体을送信) |
なし (サンドボックス内部で精製後に返却) |
| LLM往復回数 |
N回の順次往復 |
N回の順次往復 |
1回 (多段階スクリプトを一括実行) |
| タスク完了レイテンシ |
基準点 |
サーバーシリアライズのオーバーヘッド発生 |
平均 48.5% 減少 |
| トークン削減率 |
0% (基準) |
0% |
61%–98.7% 削減 |
Anthropicチームの内部分析データも類似した結果を示しています。大容量ファイルの分析作業をコード実行方式に切り替えたところ、トークン消費量が最大98.7%まで減少しました。LLMとツールの間を行き来する往復回数が1回に減ることで、レイテンシも半分近くに短縮されます。
LLMに与えるシステムプロンプトには、CLIテキストパイプラインの制約条件を明確に設定する必要があります。
`text
You operate inside a sandboxed Linux Bash environment.
To process data files or API responses, follow these constraints:
- NEVER output raw bulk data to stdout. Pipe large JSON or CSV outputs through jq, awk, or grep.
- Always inspect data structure first using head -n 5 or jq 'keys'.
- Perform aggregate operations (SUM, COUNT, GROUP BY) using bash utilities or python scripts inside the sandbox, and print ONLY the final summary result.
- Chain multiple operations into a single bash script execution to minimize inference turns.
`
10万行のCSVを分析する際、生のデータをコンテキストに放り込むのはお金をドブに捨てているようなものです。head -n 5 で構造を把握し、サンドボックス内で awk や python で集計させてから、最終結果の1行だけをモデルに返却させるよう誘導してください。
コマンドのハルシネーションとCLI未インストールエラーの補正
エージェントにBashを任せると、100% Exit Code 127(Command Not Found)や誤ったオプションフラグを吐き出します。
エージェント評価フレームワークPASTEの分析チームのレポートによると、コマンド実行が失敗した際にセッションをすぐに終了せず、自動補正ループを通過させたところ、実行失敗率が5%未満に低下しました。
`python
import re
from typing import Callable, Optional
class SelfHealingBashRunner:
def init(self, sandbox: EphemeralBashSandbox, llm_repair_fn: Callable[[str, str], str]):
self.sandbox = sandbox
self.llm_repair_fn = llm_repair_fn
self.max_retries = 3
def run_with_healing(self, initial_cmd: str) -> tuple[bool, str]:
current_cmd = initial_cmd
for attempt in range(self.max_retries):
exit_code, stdout, stderr = self.sandbox.execute_command(current_cmd)
if exit_code == 0:
return True, stdout
if exit_code == 127 or "command not found" in stderr.lower():
missing_binary = self._extract_missing_command(stderr)
if missing_binary:
install_success = self._try_install_package(missing_binary)
if install_success:
continue
repair_prompt = (
f"The executed Bash command failed.\n"
f"Failed Command: {current_cmd}\n"
f"Exit Code: {exit_code}\n"
f"Stderr Output: {stderr}\n"
f"Stdout Output: {stdout}\n"
f"Analyze the error. Return ONLY a corrected single-line Bash command to fulfill the objective."
)
current_cmd = self.llm_repair_fn(stderr, repair_prompt).strip()
return False, f"Failed after {self.max_retries} attempts. Last Stderr: {stderr}"
def _extract_missing_command(self, stderr: str) -> Optional[str]:
match = re.search(r"([a-zA-Z0-9_-]+):\s*command not found", stderr) or re.search(r"command not found:\s*([a-zA-Z0-9_-]+)", stderr)
return match.group(1) if match else None
def _try_install_package(self, binary_name: str) -> bool:
install_cmd = f"apt-get update && apt-get install -y {binary_name} || pip install {binary_name}"
exit_code, _, _ = self.sandbox.execute_command(install_cmd)
return exit_code == 0
`
エラー修復モジュールの動作フローはシンプルです。
- Exit Codeが127の場合、正規表現で必要なバイナリ名を抽出し、
apt-get や pip でリアルタイムにインストールします。
- 構文エラーや誤ったフラグの問題であれば、発生した
stderr メッセージをLLMに再度渡して自ら修正コードを作成させます。
- 検証済みのスクリプトはコンテナ内部の
/usr/local/bin/ に実行権限(chmod +x)を付与して保存します。次回からはこのバイナリを直接呼び出すだけで済むため、トークンを再消費する必要がなくなります。
ファイルの競合(コンカレンシー)の防止
サンドボックス内部のスクリプトが無限ループに陥ると、セッション全体がロックされます。階層的なタイムアウトを設定する必要があります。通常のコマンドは30秒、パッケージインストールは60秒、セッション全体は300秒で切り捨てるような形です。タイムアウトを超過した場合、モニタリングスレッドが該当プロセスのPIDに SIGKILL を送信し、問題のプロセスだけを綺麗に強制終了させるべきです。
複数のエージェントインスタンスが同じ共有ボリューム上のファイルにアクセスする際に生じるレースコンディション(競合状態)も問題です。単に open(path, 'w') を実行すると、ロックを取得する前にファイルが0バイトに吹き飛んでしまいます。POSIXカーネルレベルの fcntl.flock システムコールに基づくロックが不可欠です。
`python
import fcntl
import os
import time
from contextlib import contextmanager
class SafeFileLockTimeout(Exception):
pass
@contextmanager
def safe_file_lock(lock_file_path: str, timeout: float = 10.0, poll_interval: float = 0.05):
lock_dir = os.path.dirname(os.path.abspath(lock_file_path))
if lock_dir:
os.makedirs(lock_dir, exist_ok=True)
fd = os.open(lock_file_path, os.O_RDWR | os.O_CREAT, 0o666)
start_time = time.time()
try:
while True:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except (OSError, IOError):
if time.time() - start_time >= timeout:
raise SafeFileLockTimeout(
f"Timed out after {timeout} seconds waiting for lock on: {lock_file_path}"
)
time.sleep(poll_interval)
yield fd
finally:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
except (OSError, IOError):
pass
os.close(fd)
`
os.open を os.O_RDWR | os.O_CREAT フラグで開くことがポイントです。ロックを取得する前にファイルが切り詰められる(空になる)現象を防いでくれます。その後 fcntl.LOCK_EX | fcntl.LOCK_NB で非同期ロックを試み、タイムアウトに達した場合は例外をスローして待機スレッドが無期限に停止する状況を回避します。