Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

エンタープライズ環境でLangGraphやAutoGenベースのエージェントを実務に投入した経験のあるAIエンジニアなら、誰もが一度はヒヤッとするような経験をしたことがあるはずです。Claude Code CLIをバックグラウンドプロセスとして実行しておいたところ、たった一つのコードの欠陥のせいでツールを勝手に再呼び出しし続け、わずか10分で数 safe百ドル分のトークンを消費してしまった、というような状況です。

単にプロンプトを上手く書くだけでは解決できる問題ではありません。バックグラウンド制御の失敗、ノード間の無限待機、ブラウザのメモリクラッシュといった、実際の運用環境で発生するシステムレベルのボトルネックを解消する必要があります。本記事では、プロダクションパイプラインですぐに活用できる4つの対応コードとアーキテクチャをまとめました。

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止

AnthropicのClaude 3 Opusのようなモデルを対話なしで動作する非同期モード（-p フラグ）で実行すると、テスト失敗時にユーザーの介入なしで即座にツールを再呼び出しします。この際、無限ループに陥るとマルチステップの推論履歴が蓄積され続け、毎ターン膨大なトークンを消費することになります。プロセスが終了せずシステムメモリに残ってしまうゾンビプロセスも、この時に発生します。

この現象を防ぐには、CLIレベルで上限を設定し、スクリプト側でプロセスを強制終了できるラッパーが必要となります。

CLI呼び出し時に --max-budget-usd フラグで1回の実行あたりの支出上限を指定し、--max-turns でツールの呼び出し回数を制限します。
--allowedTools オプションで Read、Grep、Glob などの読み取り専用ツールのみを許可し、--bare フラグを付与してプラグインロードのオーバーヘッドを排除します。
Pythonの asyncio.subprocess を使用し、一定時間が経過するとプロセスをKillするラッパークラスを組み込みます。

`python
import asyncio
import os
from typing import Any, Dict

class ClaudeCodeWrapper:
def init(self, max_budget_usd: float = 0.50, max_turns: int = 5, timeout_seconds: float = 120.0):
self.max_budget_usd = max_budget_usd
self.max_turns = max_turns
self.timeout_seconds = timeout_seconds

async def execute_validation(self, prompt: str, target_dir: str) -> Dict[str, Any]:
    cmd = [
        "claude", "-p", prompt,
        "--max-budget-usd", str(self.max_budget_usd),
        "--max-turns", str(self.max_turns),
        "--allowedTools", "Read", "Grep", "Glob", "Bash(pytest *)",
        "--add-dir", target_dir,
        "--bare"
    ]
    env = os.environ.copy()
    env["CLAUDE_CODE_SIMPLE"] = "1"

    try:
        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=target_dir, env=env
        )
        try:
            stdout_data, stderr_data = await asyncio.wait_for(process.communicate(), timeout=self.timeout_seconds)
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            return {"status": "TIMEOUT_EXCEEDED", "exit_code": -1, "error": f"Timeout after {self.timeout_seconds}s"}

        if process.returncode != 0:
            return {"status": "CLI_ERROR", "exit_code": process.returncode, "error": stderr_data.decode("utf-8")}

        return {"status": "SUCCESS", "exit_code": 0, "raw_output": stdout_data.decode("utf-8")}
    except Exception as e:
        return {"status": "WRAPPER_EXCEPTION", "exit_code": -2, "error": str(e)}

このラッパーを適用すれば、検証1回あたりのコストが最大$0.50を超えることはありません。ループの混迷による予期せぬ高額請求を遮断することができます。

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

LangGraphのデフォルトの RetryPolicy はネットワークエラー対応用です。LLMがロジックを誤って作成し検証に失敗するような場合、このデフォルトポリシーは正常に機能しません。リトライ上限を超えた瞬間に GraphRecursionError をスローし、システム全体が停止してしまいます。

同一のエラーが繰り返し発生した場合は実行を中断し、制御権を安全に受け渡すサーキットブレーカー（Circuit Breaker）をState内部に構築する必要があります。

Stateに verification_attempts カウンタと last_error_signature 変数を追加します。
1回失敗して2回目のリトライに入る際は、モデルを比較的軽量なモデル（Claude 3.5 Haikuなど）に切り替えてルーティング時間を短縮します。
同一の失敗が2回を超えた場合は、Human-in-the-loop ノードに制御権を渡し、タスクを安全に上位スレッドへ引き継ぎます。

`python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic

class AgentGraphState(TypedDict):
task_prompt: str
verification_attempts: int
last_error_signature: str
verification_status: Literal["PENDING", "PASSED", "FAILED", "CIRCUIT_BROKEN"]

def verification_node(state: AgentGraphState) -> dict:
attempts = state.get("verification_attempts", 0) + 1
model_name = "claude-3-5-haiku-20241022" if attempts >= 2 else "claude-3-opus-20240229"
llm = ChatAnthropic(model=model_name, temperature=0.0)

return {"verification_attempts": attempts, "last_error_signature": "SyntaxError", "verification_status": "FAILED"}

def circuit_breaker_router(state: AgentGraphState) -> str:
if state.get("verification_status") == "PASSED":
return "proceed"
if state.get("verification_attempts", 0) >= 2:
return "trigger_fallback"
return "retry"

def human_in_the_loop_node(state: AgentGraphState) -> dict:
return {"verification_status": "CIRCUIT_BROKEN"}

builder = StateGraph(AgentGraphState)
builder.add_node("verify", verification_node)
builder.add_node("hitl_fallback", human_in_the_loop_node)
builder.add_conditional_edges("verify", circuit_breaker_router, {
"proceed": END, "retry": "verify", "trigger_fallback": "hitl_fallback"
})
builder.add_edge("hitl_fallback", END)
graph_app = builder.compile()

連続して2回失敗した時点で無限ループを断ち切り、すぐに待機状態に移行するため、エージェントがデッドロックに陥ってプロセス全体が停止する現象を防ぐことができます。

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

スクリーンショットの取得やDOM構造の検証を行うノードでは、Chrome Headless Shellを実行します。問題は、このコードがDockerやGitHub Actions Runner環境に入った瞬間にChromiumレンダラーが落ちてしまう点です。Dockerのデフォルト共有メモリ（/dev/shm）容量は64MBしかなく、画面をキャプチャする際に Failed to reserve shared memory エラーが発生します。

コンテナ環境でブラウザを立ち上げる際は、以下の設定を合わせる必要があります。

Docker実行時に --shm-size=2g オプションで共有メモリを増やし、Chromiumのオプションに --disable-dev-shm-usage および --no-sandbox フラグを追加します。
実行ごとのクッキーやローカルストレージの衝突を避けるため、--user-data-dir=/tmp/session_$RUN_ID のような独立したディレクトリをセッションごとに生成します。
CI/CDワークフローにアーティファクト保存ステップを組み込み、失敗原因を後から確認できるように構成します。

`yaml
name: Agent Background Verification Pipeline
on: [push]

jobs:
headless-browser-validation:
runs-on: ubuntu-latest
container:
image: node:20-buster
options: --shm-size=2g --user root
steps:
- uses: actions/checkout@v4
- name: Install Chrome
run: |
apt-get update && apt-get install -y wget gnupg
wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add -
sh -c 'echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list'
apt-get update && apt-get install -y google-chrome-stable --no-install-recommends
- name: Run Agent Verification
env:
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} SESSION_STORAGE_DIR: "$ {{ github.workspace }}/artifacts/sessions/${{ github.run_id }}"
run: |
mkdir -p $SESSION_STORAGE_DIR npx ts-node src/cli_validation_entry.ts --user-data-dir="$ SESSION_STORAGE_DIR"
- uses: actions/upload-artifact@v4
if: always()
with:
name: validation-artifacts-${{ github.run_id }}
path: ${{ github.workspace }}/artifacts/

メモリ領域を拡張し、セッションを完全に分離することで、バックグラウンド環境であってもブラウザ検証ノードがクラッシュすることなく最後まで正常に動作します。

4. メインStateを汚さないログモニタリングの分離

LangGraphのメインStateは、すべてのノードが共有して使用するメモリ空間です。検証ノードが取得した長いHTMLコードや実行ログ全体をそのままStateに流し込んでしまうと、次のターンのLLM呼び出し時にコンテキストウィンドウがあっという間に圧迫されます。また、DBへの保存速度も著しく低下します。

大容量データは外部ファイルへ出力し、Stateには参照パスのみを保持する手法を採用すべきです。

メインStateには validation_status と結果ファイルのパスを保持する artifact_ref_path のみを残します。
検証ノードが出力する数MB単位のテキストログは、ローカルディレクトリやS3にJSONファイルとして保存します。
トレーサビリティ確保のため、LangSmithの @traceable デコレータを付与してログの流れを連携させます。

`python
import json, os, time
from typing import TypedDict, Dict, Any
from langsmith import traceable

class LightMainState(TypedDict):
session_id: str
target_component: str
validation_status: str
artifact_ref_path: str

@traceable(run_type="llm", name="ClaudeCodeValidationSkill", tags=["claude-code-cli", "isolated-node"])
def run_context_free_validation(state: LightMainState) -> Dict[str, Any]:
session_id = state["session_id"]
raw_execution_stdout = "DUMP LOG DATA " * 10000

artifact_dir = f"./artifacts/{session_id}"
os.makedirs(artifact_dir, exist_ok=True)
artifact_full_path = os.path.join(artifact_dir, f"artifact_{int(time.time())}.json")

with open(artifact_full_path, "w", encoding="utf-8") as f:
    json.dump({"session_id": session_id, "stdout": raw_execution_stdout}, f, indent=2)

return {
    "validation_status": "PASSED",
    "artifact_ref_path": artifact_full_path
}

これにより、LLMのプロンプトに不要なログが混入するのを防ぐことができ、問題が発生した際も指定されたパスのファイルを確認するだけで済むため、原因究明が格段にスムーズになります。

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止

この現象を防ぐには、CLIレベルで上限を設定し、スクリプト側でプロセスを強制終了できるラッパーが必要となります。

CLI呼び出し時に --max-budget-usd フラグで1回の実行あたりの支出上限を指定し、--max-turns でツールの呼び出し回数を制限します。
--allowedTools オプションで Read、Grep、Glob などの読み取り専用ツールのみを許可し、--bare フラグを付与してプラグインロードのオーバーヘッドを排除します。
Pythonの asyncio.subprocess を使用し、一定時間が経過するとプロセスをKillするラッパークラスを組み込みます。

`python
import asyncio
import os
from typing import Any, Dict

async def execute_validation(self, prompt: str, target_dir: str) -> Dict[str, Any]:
    cmd = [
        "claude", "-p", prompt,
        "--max-budget-usd", str(self.max_budget_usd),
        "--max-turns", str(self.max_turns),
        "--allowedTools", "Read", "Grep", "Glob", "Bash(pytest *)",
        "--add-dir", target_dir,
        "--bare"
    ]
    env = os.environ.copy()
    env["CLAUDE_CODE_SIMPLE"] = "1"

    try:
        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=target_dir, env=env
        )
        try:
            stdout_data, stderr_data = await asyncio.wait_for(process.communicate(), timeout=self.timeout_seconds)
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            return {"status": "TIMEOUT_EXCEEDED", "exit_code": -1, "error": f"Timeout after {self.timeout_seconds}s"}

        if process.returncode != 0:
            return {"status": "CLI_ERROR", "exit_code": process.returncode, "error": stderr_data.decode("utf-8")}

        return {"status": "SUCCESS", "exit_code": 0, "raw_output": stdout_data.decode("utf-8")}
    except Exception as e:
        return {"status": "WRAPPER_EXCEPTION", "exit_code": -2, "error": str(e)}

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

Stateに verification_attempts カウンタと last_error_signature 変数を追加します。
1回失敗して2回目のリトライに入る際は、モデルを比較的軽量なモデル（Claude 3.5 Haikuなど）に切り替えてルーティング時間を短縮します。
同一の失敗が2回を超えた場合は、Human-in-the-loop ノードに制御権を渡し、タスクを安全に上位スレッドへ引き継ぎます。

`python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic

class AgentGraphState(TypedDict):
task_prompt: str
verification_attempts: int
last_error_signature: str
verification_status: Literal["PENDING", "PASSED", "FAILED", "CIRCUIT_BROKEN"]

return {"verification_attempts": attempts, "last_error_signature": "SyntaxError", "verification_status": "FAILED"}

def human_in_the_loop_node(state: AgentGraphState) -> dict:
return {"verification_status": "CIRCUIT_BROKEN"}

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

コンテナ環境でブラウザを立ち上げる際は、以下の設定を合わせる必要があります。

Docker実行時に --shm-size=2g オプションで共有メモリを増やし、Chromiumのオプションに --disable-dev-shm-usage および --no-sandbox フラグを追加します。
実行ごとのクッキーやローカルストレージの衝突を避けるため、--user-data-dir=/tmp/session_$RUN_ID のような独立したディレクトリをセッションごとに生成します。
CI/CDワークフローにアーティファクト保存ステップを組み込み、失敗原因を後から確認できるように構成します。

`yaml
name: Agent Background Verification Pipeline
on: [push]

4. メインStateを汚さないログモニタリングの分離

大容量データは外部ファイルへ出力し、Stateには参照パスのみを保持する手法を採用すべきです。

メインStateには validation_status と結果ファイルのパスを保持する artifact_ref_path のみを残します。
検証ノードが出力する数MB単位のテキストログは、ローカルディレクトリやS3にJSONファイルとして保存します。
トレーサビリティ確保のため、LangSmithの @traceable デコレータを付与してログの流れを連携させます。

`python
import json, os, time
from typing import TypedDict, Dict, Any
from langsmith import traceable

class LightMainState(TypedDict):
session_id: str
target_component: str
validation_status: str
artifact_ref_path: str

artifact_dir = f"./artifacts/{session_id}"
os.makedirs(artifact_dir, exist_ok=True)
artifact_full_path = os.path.join(artifact_dir, f"artifact_{int(time.time())}.json")

with open(artifact_full_path, "w", encoding="utf-8") as f:
    json.dump({"session_id": session_id, "stdout": raw_execution_stdout}, f, indent=2)

return {
    "validation_status": "PASSED",
    "artifact_ref_path": artifact_full_path
}

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

Related Video

Anthropicがグラフエンジニアリング最大の欠陥を修正

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

4. メインStateを汚さないログモニタリングの分離

Comments (0)

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

4. メインStateを汚さないログモニタリングの分離

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

1. -p オプション実行時のトークン暴走とプロセスゾンビ化の防止

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

4. メインStateを汚さないログモニタリングの分離

Comments (0)

Claude Code CLIのコスト暴走と無限ループを防ぐエージェントパイプライン構築法

1. -p オプション実行時のトークン暴走とプロセスゾンビ化の防止

2. グラフノードの無限リトライを遮断するサーキットブレーカーの実装

3. CI/CDおよびDocker内部のChrome Headless Shellメモリ隔離

4. メインStateを汚さないログモニタリングの分離

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止

1. `-p` オプション実行時のトークン暴走とプロセスゾンビ化の防止