Unbounded Consumption 対策：AI エージェントにステップ数・トークン・費用の上限を付け、呼ぶ前に止める

エージェントは 1 回の依頼で、モデルを何度も呼び、ツールを何度も実行します。ループに入る、検索を何十回も繰り返す、推論モデルが長く考え込む。どれも 1 回の API 呼び出しを見ているだけでは異常に見えませんが、積み上がると費用・外部 API の利用枠・処理時間を使い切ります。

この記事では、エージェントの 1 実行と利用者ごとに上限を付けるガードを Python で書き、ループ・大量のツール呼び出し・推論の長期化・巨大なツール結果・時間切れ・同じ利用者の繰り返しで、どの上限がどこで止めるかを確かめます。主要なエージェント用 SDK に既定で入っている上限の値も、インストールしたソースで確認します。

Unbounded Consumption は 2026 年版で 6 位に上がった

OWASP Top 10 for LLM Applications の 2026 年版では、Unbounded Consumption が LLM06 になり、2025 年版の 10 位から 4 つ上がりました。Cloud Security Alliance に転載された解説記事は、上がった理由として、長い推論が短い入力から大量のトークンを使うことと、エージェントが 1 つの依頼を繰り返しのモデル呼び出しとツール実行に変えることを挙げています。この項目には、費用を使わせる攻撃（Denial of Wallet）、サービスの停止、モデルの抽出が含まれます。

同じ記事は、要求の回数で数える従来のレート制限では、こうした処理の重さを測れないと指摘しています。1 リクエストの重さが数十倍違うので、「1 分に何回」だけでは足りません。

攻撃として意図的に重くする研究もあります。OverThink（Kumar ら、arXiv:2502.02542）は、RAG などで推論モデルが読む公開コンテンツにおとりの推論問題を埋め込み、答えは正しいまま推論トークンだけを増やす攻撃です。論文は o1・o1-mini・o3-mini・DeepSeek R1 を対象に、FreshQA と SQuAD で最大 46 倍の遅延を報告しています。おとりの問題そのものは有害な内容ではないので、安全性のガードレールでは引っかかりません。対策としては、検索結果の絞り込みや言い換えが論文中で検討されています。

ここから分かるのは、費用の上限は内容の検査とは別の層として持つ必要があるということです。

何に上限を付けるか

1 実行（エージェントが 1 つの依頼を処理する間）と、利用者ごとの期間の 2 つの単位で考えます。

単位	上限	止めるもの
1 回のモデル呼び出し	出力トークン（推論を含む）	推論の長期化、長い出力の誘発
1 実行	ステップ数・合計トークン・費用・経過時間	ループ、考え込み、遅いツールの連続
1 実行 × ツール	ツールごとの回数、同じ引数での繰り返し	検索の大量実行、同じ呼び出しのループ
1 実行 × ツール結果	コンテキストに戻す文字数	巨大なページやファイルの取り込み
利用者 × 1 日	費用	同じ人が重い依頼を繰り返す、盗まれたキー

ポイントは 2 つあります。

• 呼ぶ前に止める: 呼んだ後に合計を見て止めると、最後の 1 回で上限を大きく超えます。呼ぶ前に「出力が上限いっぱいまで出た場合」の費用を見積もり、収まらなければ呼びません
• 推論トークンを数える: 推論モデルの推論トークンは画面に出ませんが、出力トークンとして課金されます。OpenAI の Python SDK（openai 3.46.0）の型定義では、Responses API の max_output_tokens は表示される出力と推論トークンの両方を含む上限で、usage.output_tokens_details.reasoning_tokens が出力トークンの内訳として返ります

上限を持つガードを書く

実行環境は Python 3.13.18、標準ライブラリだけです。価格は説明用の仮の値（100 万トークンあたり入力 3 ドル、出力 15 ドル）です。実際の価格はモデルごとに違うので、使うモデルの料金表に合わせてください。

budget.py

import json
import threading
import time
from collections import Counter
from dataclasses import dataclass, field

# 価格は説明用の仮の値（100 万トークンあたりの米ドル）
PRICE = {"input": 3.00, "output": 15.00}


def cost_usd(input_tokens: int, output_tokens: int) -> float:
    return (input_tokens * PRICE["input"] + output_tokens * PRICE["output"]) / 1_000_000


class BudgetExceeded(Exception):
    def __init__(self, reason: str):
        super().__init__(reason)
        self.reason = reason


@dataclass
class Limits:
    max_steps: int = 8                  # モデル呼び出しの回数
    max_tokens: int = 60_000            # 入力 + 出力（推論トークンを含む）
    max_cost_usd: float = 0.30
    max_wall_seconds: float = 120.0
    max_output_tokens_per_call: int = 4_000   # API の上限指定にもそのまま渡す
    max_calls_per_tool: dict = field(default_factory=lambda: {"web_search": 5, "send_email": 1})
    max_same_call: int = 2              # 同じツールを同じ引数で呼べる回数
    max_tool_result_chars: int = 8_000  # ツール結果をコンテキストに戻すときの上限


class RunBudget:
    """1 回の依頼（エージェントの 1 実行）に付ける上限"""

    def __init__(self, limits: Limits, clock=time.monotonic):
        self.l = limits
        self.clock = clock
        self.started = clock()
        self.steps = 0
        self.tokens = 0
        self.cost = 0.0
        self.tool_calls = Counter()
        self.same_calls = Counter()

    def before_model_call(self, input_tokens: int) -> float:
        """呼ぶ前に、最悪の場合（出力が上限いっぱい）でも枠に収まるかを見る"""
        if self.clock() - self.started > self.l.max_wall_seconds:
            raise BudgetExceeded("wall_time")
        if self.steps + 1 > self.l.max_steps:
            raise BudgetExceeded("steps")
        worst_tokens = input_tokens + self.l.max_output_tokens_per_call
        if self.tokens + worst_tokens > self.l.max_tokens:
            raise BudgetExceeded("tokens")
        worst_cost = cost_usd(input_tokens, self.l.max_output_tokens_per_call)
        if self.cost + worst_cost > self.l.max_cost_usd:
            raise BudgetExceeded("cost")
        return worst_cost  # 利用者単位の枠に仮押さえする額

    def after_model_call(self, usage: dict) -> float:
        """API が返した usage（実測値）で精算する"""
        out = usage["output_tokens"]  # 推論トークンを含む値を使う
        c = cost_usd(usage["input_tokens"], out)
        self.steps += 1
        self.tokens += usage["input_tokens"] + out
        self.cost += c
        return c

    def before_tool_call(self, name: str, args: dict) -> None:
        cap = self.l.max_calls_per_tool.get(name)
        if cap is not None and self.tool_calls[name] + 1 > cap:
            raise BudgetExceeded(f"tool_calls:{name}")
        key = (name, json.dumps(args, sort_keys=True, ensure_ascii=False))
        if self.same_calls[key] + 1 > self.l.max_same_call:
            raise BudgetExceeded(f"repeated_call:{name}")
        self.tool_calls[name] += 1
        self.same_calls[key] += 1

    def clip_tool_result(self, text: str) -> str:
        n = self.l.max_tool_result_chars
        if len(text) <= n:
            return text
        return text[:n] + f"\n[省略: 残り {len(text) - n} 文字]"


class UserLedger:
    """利用者ごとの 1 日の上限。並行実行でも超えないよう、呼ぶ前に仮押さえする"""

    def __init__(self, daily_limit_usd: float):
        self.limit = daily_limit_usd
        self.spent: dict[str, float] = {}
        self.reserved: dict[str, float] = {}
        self.lock = threading.Lock()

    def reserve(self, user: str, amount: float) -> None:
        with self.lock:
            used = self.spent.get(user, 0.0) + self.reserved.get(user, 0.0)
            if used + amount > self.limit:
                raise BudgetExceeded("user_daily")
            self.reserved[user] = self.reserved.get(user, 0.0) + amount

    def settle(self, user: str, reserved: float, actual: float) -> None:
        with self.lock:
            self.reserved[user] -= reserved
            self.spent[user] = self.spent.get(user, 0.0) + actual

設計で意識したことを補足します。

• 仮押さえと精算: 利用者単位の枠は、呼ぶ前に最悪の場合の額を仮押さえし、API が返した実測値で精算します。同じ利用者のエージェントが並行して動いても、仮押さえの合計で判定するので枠を超えません。本番では UserLedger を Redis やデータベースの原子的な更新に置き換えます
• 同じ呼び出しの判定: 引数をキー順に並べた JSON にしてから数えます。キーの順番が違うだけの呼び出しを別物として数えないためです
• ツール結果の切り詰め: 巨大な結果をそのまま戻すと、次の呼び出しの入力トークンが膨らみ、以後のすべてのステップが重くなります。戻す前に切り詰め、省略したことをモデルに伝えます

モデルの出力を固定値で与えて、どこで止まるかを確かめる

LLM は呼ばず、モデルの 1 回ごとの出力トークン数と呼び出すツールをスクリプトとして固定値で与えます。入力トークンは実際に積み上がったコンテキスト長を使います。

run_agent.py

"""モデルの応答は固定値（スクリプト）で与え、上限の効き方だけを確かめる"""
from budget import BudgetExceeded, Limits, RunBudget, UserLedger


def run(user, script, ledger, limits=None, clock=None, tools=None):
    b = RunBudget(limits or Limits(), **({"clock": clock} if clock else {}))
    tools = tools or {}
    context_tokens = 1_500
    try:
        for reply in script:
            hold = b.before_model_call(context_tokens)
            ledger.reserve(user, hold)
            # ここで本来は LLM API を max_output_tokens 付きで呼ぶ
            # 入力は実際のコンテキスト長、出力はスクリプトの固定値を usage とみなす
            usage = {"input_tokens": context_tokens, "output_tokens": reply["out"]}
            actual = b.after_model_call(usage)
            ledger.settle(user, hold, actual)
            context_tokens += reply["out"]
            call = reply.get("tool")
            if call is None:
                return "done", b
            b.before_tool_call(call["name"], call["args"])
            result = tools.get(call["name"], lambda a: "ok")(call["args"])
            clipped = b.clip_tool_result(result)
            if clipped != result:
                print(f"  clip: {len(result)} -> {len(clipped)} 文字")
            result = clipped
            context_tokens += len(result) // 2  # 説明用の大まかな換算
        return "script_end", b
    except BudgetExceeded as e:
        return f"stopped:{e.reason}", b


normal = [
    {"out": 300, "tool": {"name": "web_search", "args": {"q": "請求書 期限"}}},
    {"out": 250, "tool": {"name": "send_email", "args": {"to": "[e-mail]"}}},
    {"out": 200},
]
loop = [{"out": 120, "tool": {"name": "web_search", "args": {"q": "同じ質問"}}}] * 20
fanout = [{"out": 120, "tool": {"name": "web_search", "args": {"q": f"q{i}"}}} for i in range(20)]
overthink = [{"out": 3_900, "tool": {"name": "web_search", "args": {"q": f"d{i}"}}} for i in range(20)]
huge_result = [
    {"out": 100, "tool": {"name": "fetch", "args": {"url": "https://example.com/big"}}},
    {"out": 100},
]

ledger = UserLedger(daily_limit_usd=1.00)
for name, script in [("normal", normal), ("loop", loop), ("fanout", fanout), ("overthink", overthink)]:
    status, b = run("alice", script, ledger)
    print(f"{name:10s} {status:28s} steps={b.steps} tokens={b.tokens:6d} cost=${b.cost:.4f}")

status, b = run("bob", huge_result, ledger, tools={"fetch": lambda a: "x" * 500_000})
print(f"{'huge':10s} {status:28s} steps={b.steps} tokens={b.tokens:6d} cost=${b.cost:.4f}")

# 時間の上限：時計を差し替えて、1 ステップごとに 50 秒進める
t = [0.0]
def fake_clock():
    t[0] += 50.0
    return t[0]
status, b = run("carol", fanout, ledger, clock=fake_clock)
print(f"{'slow':10s} {status:28s} steps={b.steps}")

# 利用者単位の上限：同じ人が重い依頼を繰り返す
print("alice spent", round(ledger.spent["alice"], 4))
for i in range(6):
    status, b = run("dave", overthink, ledger)
    print(f"dave#{i}     {status:28s} cost=${b.cost:.4f} spent=${ledger.spent['dave']:.4f}")

実行結果です。

normal     done                         steps=3 tokens=  6103 cost=$0.0273
loop       stopped:repeated_call:web_search steps=3 tokens=  5223 cost=$0.0200
fanout     stopped:tool_calls:web_search steps=6 tokens= 11535 cost=$0.0432
overthink  stopped:cost                 steps=3 tokens= 27903 cost=$0.2241
  clip: 500000 -> 8019 文字
huge       done                         steps=2 tokens=  7309 cost=$0.0243
slow       stopped:wall_time            steps=2
alice spent 0.3147
dave#0     stopped:cost                 cost=$0.2241 spent=$0.2241
dave#1     stopped:cost                 cost=$0.2241 spent=$0.4482
dave#2     stopped:cost                 cost=$0.2241 spent=$0.6723
dave#3     stopped:cost                 cost=$0.2241 spent=$0.8964
dave#4     stopped:user_daily           cost=$0.0630 spent=$0.9594
dave#5     stopped:user_daily           cost=$0.0000 spent=$0.9594

読み取れることを整理します。

• ループ（同じ検索を繰り返す）は、ステップ数の上限 8 に届く前に、同じ引数の 3 回目で止まりました。ステップ数だけで止めると、上限までの 8 回分を毎回払うことになります
• 大量のツール呼び出し（引数を変えて検索を続ける）は、同じ引数の判定をすり抜けますが、ツールごとの回数 5 で止まりました
• 推論の長期化（1 回あたり 3,900 トークンを出力）は 3 ステップ目の後、4 回目を呼ぶ前に費用の上限で止まりました。止まった時点の費用は 0.2241 ドルで、上限 0.30 ドルを超えていません。合計を呼んだ後に見る方式だと、4 回目を呼んでから気づくので上限を超えます
• 巨大なツール結果は 50 万文字から約 8,000 文字に切り詰められ、次のステップの入力が膨らみませんでした。切り詰めなければ約 25 万トークン（この記事の大まかな換算）がコンテキストに入り、次の呼び出しの前にトークンの上限で止まります。止まるので被害は小さいものの、依頼は失敗します
• 時間は、1 ステップに 50 秒かかる設定で、3 回目を呼ぶ前に 120 秒の上限で止まりました
• 同じ利用者の繰り返しは、1 回ずつは実行単位の上限内でも、5 回目で 1 日の上限に届きました。4 回目の途中で止まっているのは、仮押さえが残り枠に入らなかったためで、仮押さえ込みでも累計は 1 ドルを超えていません

SDK の既定の上限を確かめる：版によって変わる

自分で書かなくても、主要なエージェント用 SDK にはステップ数の上限があります。ただし既定値は SDK ごとに違い、版によって変わります。2026 年 11 月 7 日に PyPI からインストールした版のソースで確認した値です。

SDK（版）	設定	既定値	超えたとき
openai-agents（0.24.0）	Runner.run(..., max_turns=)	10	MaxTurnsExceeded 例外
langgraph（1.2.20）	config={"recursion_limit": ...}	10007	GraphRecursionError 例外
claude-agent-sdk（0.2.179）	ClaudeAgentOptions(max_turns=, max_budget_usd=)	どちらも None（指定なし）	費用超過時は error_max_budget_usd の結果

LangGraph の再帰上限は、以前の解説では 25 と書かれていることが多い値です。手元の 1.2.20 では langgraph/_internal/_config.py で 10007（環境変数 LANGGRAPH_DEFAULT_RECURSION_LIMIT で変更可）になっていました。終了条件のないループを持つグラフで実際に確かめた結果です。

lg_check.py

from typing import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.errors import GraphRecursionError

class S(TypedDict):
    n: int

def step(s: S) -> S:
    return {"n": s["n"] + 1}

g = StateGraph(S)
g.add_node("step", step)
g.add_edge(START, "step")
g.add_edge("step", "step")  # 終了条件のないループ
app = g.compile()

for cfg in ({}, {"recursion_limit": 25}):
    try:
        app.invoke({"n": 0}, config=cfg)
    except GraphRecursionError as e:
        print(cfg or "既定値", "->", type(e).__name__, str(e).splitlines()[0][:70])

既定値 -> GraphRecursionError Recursion limit of 10007 reached without hitting a stop condition. You
{'recursion_limit': 25} -> GraphRecursionError Recursion limit of 25 reached without hitting a stop condition. You ca

既定のままだと、ループに入ったグラフは 1 万回を超えてノードを実行します。ノードの中でモデルを呼んでいれば、その回数分の費用がかかります。SDK の既定値に頼らず、アプリ側で必ず明示的に指定し、SDK を更新したら値を確認するのが安全です。

また、SDK のステップ数の上限が数えるのはステップの回数だけで、1 ステップの重さは見ていません。推論の長期化や巨大な入力には、モデル呼び出しごとの出力上限（OpenAI の Responses API なら max_output_tokens）と、上のガードのような費用の上限を組み合わせます。Responses API には組み込みツールの呼び出し回数の上限 max_tool_calls もありますが、型定義の説明によれば、これは 1 レスポンス内の組み込みツール全体の合計で、自前のツールやエージェントの実行全体には効きません。

上限に達したときの振る舞いを決めておく

上限は「止める」だけでなく、止めた後にどう見せるかまで決めておかないと、利用者が再実行を繰り返して結局同じ費用がかかります。

• 途中経過を返す: 止まった理由（ステップ数・費用・時間）と、そこまでに分かったことを返します。エラーだけを返すと、同じ依頼がそのまま再送されます
• 再開の判断を人に渡す: 上限に達した実行を続けるかどうかは、利用者または管理者が明示的に選ぶようにします。モデル自身に「上限を上げて続ける」判断をさせません
• 理由ごとに記録する: repeated_call が多いならプロンプトかツールの設計の問題、cost が多いなら上限が低すぎるか攻撃の可能性、というように原因の切り分けに使えます。利用者単位の上限に達した記録は、盗まれた API キーや乗っ取られたアカウントの手がかりにもなります
• 外部 API の待ち時間にも上限を付ける: このガードの時間判定はステップの間でしか効きません。モデル API やツールの呼び出し自体にもタイムアウトを設定します

レッドチームで上限が効くかを確かめる

上限を入れたら、効いていることを試験で確かめます。上のスクリプトと同じ形で、モデルの応答を固定値にした回帰テストとして残しておくと、設定変更や SDK の更新で上限が外れたときに気づけます。

• 同じツールを同じ引数で呼び続ける応答、引数を少しずつ変えて呼び続ける応答
• 1 回あたりの出力が出力上限ぎりぎりの応答が続くケース（推論の長期化の代わり）
• ツールが巨大な結果を返すケース、ツールの応答が遅いケース
• 同じ利用者が並行して重い依頼を送るケース（仮押さえが効いているか）

実際のモデルを使う試験では、「処理を止めずに続けて」「すべての候補を 1 件ずつ調べて」のような継続や網羅を求める依頼や、RAG の文書に長い推論を誘う内容を入れたケースで、上限のどれが先に効くかを記録します。

参考

• OWASP GenAI Security Project: 2026 Top 10 for LLM Applications の発表
• Cloud Security Alliance: The OWASP Top 10 for LLM Applications 2026: What Changed and Why It Matters
• Kumar et al.: OverThink: Slowdown Attacks on Reasoning LLMs (arXiv:2502.02542)
• openai/openai-agents-python
• langchain-ai/langgraph
• anthropics/claude-agent-sdk-python
• openai/openai-python