専門家一覧
AIエージェントのループ回数を制限するトークン・ガバナンスの自動化

AIエージェントの無限ループを防ぐ：トークン・ガバナンス自動化の実装パターン

この記事は急速に進化する技術について解説しています。最新情報は公式ドキュメントをご確認ください。

約18分で読めます
文字サイズ:
AIエージェントの無限ループを防ぐ：トークン・ガバナンス自動化の実装パターン
目次

この記事の要点

  • AIエージェントの無限ループを未然に防止
  • トークン消費の予測不能なコスト爆発を抑制
  • LangGraphやRedisを用いた具体的な実装パターン

「PoC（概念実証）では上手くいっていたAIエージェントが、本番環境で特定の入力を受けた瞬間、無限ループに陥り、一晩で数千ドルのAPIコストを溶かしてしまった」

これは決して笑い話ではなく、自律型AIを本番導入した多くの現場で報告されている深刻なインシデントです。自律型AIエージェント（Autonomous Agents）は、ReActパターンなどに代表されるように、「思考」と「行動」を繰り返してタスクを解決します。エージェントの実装パターンは日々進化しているため、最新の推奨手順は公式ドキュメントを直接確認する必要がありますが、この「繰り返し（ループ）」こそが、最大のパワーであり、同時に最大のリスク要因であるという本質は変わりません。

LLM（大規模言語モデル）は確率論的に動作するため、100回に1回、あるいは1000回に1回、予期せぬ判断を下すリスクを孕んでいます。その結果、エラー画面を延々と読み込もうとしたり、同じ検索クエリを投げ続けたりする「暴走」が始まります。

多くのエンジニアは max_iterations（最大反復回数）を設定して安心していますが、本番運用においてはそれだけでは不十分です。なぜなら、回数制限に達するまでの間に、GPT-5.2（InstantやThinking）のような長文コンテキストを扱う高機能モデルが数十回呼び出されれば、それだけで甚大な損失に直結するからです。ちなみに、GPT-4oなどの旧モデルは2026年2月に廃止されているため、本番環境では最新モデルへの確実な移行と合わせた厳密なコスト管理が急務となります。

本記事では、AIエージェントを本番環境へデプロイしようとしているプロジェクトマネージャーやテックリードの方々に向けて、「エンジニアリングとしてのガバナンス」を実装する実践的なアプローチを提示します。これは単なる倫理規定やポリシーの話ではありません。コードレベルで物理的にコストと挙動を制御し、ROIを最大化するための堅牢なミドルウェアの構築手法です。

現行バージョンのLangGraphや、メモリ管理が大幅に改善された最新のRedis（バージョン8.6.0等）、そしてベクトル検索技術を組み合わせ、エージェントの自律性を損なわずに手綱を握るための具体的なアーキテクチャの全貌に迫ります。なお、LangGraphのチェックポイントAPIなどの詳細な仕様は変更される可能性があるため、実装時は常に公式ドキュメントで最新情報を確認してください。

1. 技術的背景：なぜAIエージェントは「暴走」し、コストを浪費するのか

なぜAIエージェントは、従来のソフトウェアのように単純な while ループの条件式だけで制御しきれないのでしょうか。その根本的な原因を紐解いていきます。

自律ループの構造的リスクとAPIコールの指数関数的増加

従来のプログラムにおいて、ループの終了条件は「配列の最後まで処理したら」「変数が特定の値になったら」と極めて明確です。しかし、LLMを用いたエージェントの場合、終了条件は「LLMが『タスク完了』と判断したら」という非常に曖昧なものになります。

かつて主流だったReAct（Reasoning + Acting）パターンのような旧来のアーキテクチャでは、「思考（Thought）→ 行動（Action）→ 観察（Observation）」というサイクルをLLM自身の判断のみで回していました。しかし、最新のLangChainやLangGraphの公式ドキュメントでは、この単純なReActパターンの記述は姿を消しつつあります。現在では、ノードとエッジで明示的に状態を管理するグラフベースのアーキテクチャ（Tool Callingエージェントなど）への移行が推奨されています。

これからエージェントを構築・改修する場合は、ブラックボックス化しやすい旧来の自律ループを廃止し、LangGraphなどを用いて「どの状態で、どのツールを呼び出し、いつ終了するか」を明示的なステートマシンとして定義し直すステップを踏んでください。

なぜ旧来のパターンが非推奨になりつつあるのでしょうか。問題は、ツールの実行結果がLLMの想定と異なるときに顕著に表れます。例えば、検索結果が「該当なし」だった場合、賢いモデルなら検索ワードを変えますが、文脈やプロンプトの解釈によっては「もう一度同じ検索をする」という無意味な判断を下すことがあります。

さらに恐ろしいのがコンテキストの肥大化です。自律ループが回るたびに、「思考・行動・結果」の履歴がプロンプトに蓄積されていきます。つまり、1回目のループと10回目のループでは、1回あたりの入力トークン数が桁違いに増えるのです。

  • 1回目: 1,000トークン
  • 5回目: 5,000トークン
  • 10回目: 10,000トークン

もし10回ループして失敗した場合、単純計算でも累積で数万トークンを消費します。これが複数のユーザーによって同時に引き起こされたら、APIコストは線形ではなく、指数関数的に跳ね上がってしまいます。

従来の静的タイムアウト設定の限界

Webサーバーの設定でよくある「タイムアウト（例: 30秒）」といった対策も、AIエージェントには通用しません。LLMの生成速度や外部ツールの実行時間は変動が激しく、30秒で終わるタスクもあれば、正当な処理として3分かかるタスクも存在するからです。

単純に時間を区切ってしまうと、正常に動作している複雑なタスクまで強制終了させてしまい、ユーザー体験（UX）を著しく損ないます。逆に制限時間を長く設定すれば、暴走時の金銭的ダメージが拡大するジレンマに陥ります。

「トークン・ガバナンス」が必要な3つのレイヤー

したがって、本番運用において実装すべきは静的な制限ではなく、以下の3つのレイヤーにまたがる動的なガバナンス機構です。最新のグラフベースフレームワークへ移行した上で、これらの制御を組み込むことが不可欠となります。

  1. ハード・ガバナンス: 物理的な最大ステップ数の制限と緊急停止スイッチ（安全弁）
  2. バジェット・ガバナンス: トークン消費量に基づいた動的な予算制御（財布の紐）
  3. セマンティック・ガバナンス: 行動の意味内容や反復パターンに基づいた異常検知（監視カメラ）

これらをアプリケーションコードの中に直接書き込むのではなく、独立した制御ロジックとしてステートマシンに組み込むことが、本番運用の安定性を担保する鍵となります。

2. アーキテクチャ設計：ガバナンス・ミドルウェアの構築

具体的なシステム構成とアーキテクチャ設計を検討します。アプリケーションコードとLLMプロバイダーの間に配置する「ガバナンス・ミドルウェア」が、システム全体の要となります。

エージェント実行環境へのインターセプター導入

AIエージェントのロジック（LangChainやLangGraphで記述されたグラフ）と、実際のLLMプロバイダー（OpenAI APIなど）の間に、「ガバナンス・ミドルウェア」を配置します。

このミドルウェアは、プロキシサーバーのように振る舞うことも、Pythonのデコレータやコールバックハンドラとして実装することも可能です。LangChainの BaseCallbackHandler を拡張し、すべてのLLM呼び出しとツール実行をフックするアプローチが一般的です。

さらに、LangGraphを活用している場合、状態管理（checkpoints API）の仕組みが日々進化しています。最新の公式ドキュメントを確認し、永続化の拡張機能（例えばDynamoDBSaverなど）とミドルウェアをどのように連携させるか、プロジェクトの要件に合わせて設計することが不可欠です。

また、LLMプロバイダー側でも、古いモデルの廃止と新しいモデルへの移行が定期的に行われます。例えばOpenAIでは、レガシーモデルが順次提供終了となり、より高性能な最新の標準モデルやコーディング特化モデルへの移行が進んでいます。ミドルウェア層でこれらの最新モデルへのルーティングやトークン消費を監視することで、アプリケーション本体のコードを改修する負担を大幅に軽減できます。

トークンバケットアルゴリズムを用いた動的予算管理

APIのレート制限（Rate Limiting）でよく使われる「トークンバケットアルゴリズム」を応用します。ただし、ここでは「リクエスト回数」ではなく、「トークン量（通貨）」を管理対象とします。

  • ユーザーごとのバケット: 各ユーザー（またはセッション）には、一定期間内に使用できる「トークン予算」が割り当てられます。
  • 消費: LLMを呼び出すたびに、見積もりトークン数をバケットから減算します。利用するモデル（標準モデルか、軽量版かなど）によって単価が異なるため、動的な計算が求められます。
  • 補充: 時間経過や課金プランに応じて予算が補充されます。

このロジックをエージェントのループ内に組み込むことで、「あと何回回せるか」ではなく「あと何円使えるか」という、より経営的な視点での制御が可能になります。プロジェクトのROIを管理する上でも非常に有効なアプローチです。

状態管理DB（Redis等）によるリアルタイム監視設計

エージェントはステートフル（状態を持つ）ですが、Webサーバーはステートレス（状態を持たない）であることが一般的です。分散環境（Kubernetes上の複数のポッドなど）でガバナンスを効かせるには、状態を外部ストアで共有する必要があります。

ここで有力な選択肢となるのが Redis です。近年のアップデートではメモリ構造の最適化やパフォーマンス向上が図られており、監視基盤としての信頼性がさらに高まっています。

  • 高速性: LLMの呼び出しごとにチェックしてもレイテンシへの影響が軽微。
  • TTL（Time To Live）: セッションの有効期限管理が容易。
  • アトミック操作: 複数の並列処理が同時に予算を消費しようとした際の整合性を担保。

AWS環境などでは、LangGraphの永続化拡張としてAmazon DynamoDBを採用するケースも増えていますが、リアルタイムなトークンバケットの増減管理という点では、インメモリデータストアであるRedisの高速性が活きます。

アーキテクチャ概要図のイメージ:

  1. User Request -> Agent Runner
  2. Agent Runner -> Governance Middleware (Check Budget & Policy)
  3. Governance Middleware <-> Redis (Get/Set Token Count)
  4. Governance Middleware -> LLM API (Execute)
  5. Governance Middleware -> Log Store (Audit)

この構成により、エージェントのロジックを汚すことなく、横断的なガバナンスを適用できます。

3. 実装ステップ1：ハードリミットと「キルスイッチ」の実装

2. アーキテクチャ設計：ガバナンス・ミドルウェアの構築 - Section Image

PythonとLangGraphを用いた具体的な実装手順を解説します。エージェント開発において、まずは基本となる物理的な停止機能の組み込みが不可欠です。

LangGraphでの再帰回数制限（Recursion Limit）の設定

LangGraphを使用する場合、グラフの実行時に recursion_limit を指定することで、ノード間の遷移回数に上限を設けることができます。これは最も基本的な防御策として機能します。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 状態定義
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# グラフ構築（省略）
workflow = StateGraph(AgentState)
# ... ノードとエッジの定義 ...
app = workflow.compile()

# 実行時の制限設定
inputs = {"messages": [("user", "複雑な調査タスクを実行して")]}
config = {"recursion_limit": 10}  # 最大10ステップで強制終了

try:
    for output in app.stream(inputs, config=config):
        # 各ステップの処理
        pass
except Exception as e:
    # RecursionLimitエラーのハンドリング
    print(f"Task terminated due to recursion limit: {e}")

この設定は必須の対策ですが、これだけでは「10回までは暴走を許容する」ことになってしまいます。また、外部から即座に手動で止めたい場合には無力です。

なお、LangGraphのエコシステムでは、checkpoints APIを利用した状態の永続化機能が提供されています。上限に達して処理が中断された場合でも、その時点の状態をデータベース（Amazon DynamoDBなど）に確実に保存し、後から安全に再開や分析を行う設計を取り入れることが推奨されます。これにより、エラー発生時の原因究明や途中からの実行再開が容易になります。

緊急停止用キルスイッチのコード実装パターン

運用中、「このエージェントの挙動がおかしい」と気づいた瞬間に停止させるための「キルスイッチ」を実装します。ここでは、Redisのフラグ機能を利用したアプローチを紹介します。

特に、OpenAIのGPT-4oをはじめとする推論能力の高いモデルなどは、一度制御を失うと短時間で大量のトークンを消費する恐れがあるため、迅速に通信を遮断する仕組みが欠かせません。

import redis
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.exceptions import OutputParserException

# Redisクライアントの初期化
r = redis.Redis(host='localhost', port=6379, db=0)

class KillSwitchCallbackHandler(BaseCallbackHandler):
    def __init__(self, session_id: str):
        self.session_id = session_id

    def on_llm_start(self, serialized, prompts, kwargs):
        # LLM呼び出し直前にキルスイッチを確認
        status = r.get(f"session:{self.session_id}:status")
        if status and status.decode('utf-8') == 'killed':
            raise Exception("Kill switch activated: Process terminated by administrator.")

    def on_tool_start(self, serialized, input_str, kwargs):
        # ツール実行前にも確認
        status = r.get(f"session:{self.session_id}:status")
        if status and status.decode('utf-8') == 'killed':
            raise Exception("Kill switch activated: Process terminated by administrator.")

# 使用例
# 管理画面から 'session:12345:status' を 'killed' に設定すると、
# 次のアクション実行時に即座に例外が発生して停止する。

このハンドラをLangChainの callbacks に渡すことで、実行中のどのアクションのタイミングでも割り込んで停止させることが可能になります。

強制終了時のグレースフル・シャットダウン処理

強制終了時に単にシステムエラーを吐いて終わるのではなく、ユーザーには「処理を中断しました」と適切に伝え、そこまでのログや成果物を保存する設計が求められます。

try:
    app.invoke(inputs, config={"callbacks": [KillSwitchCallbackHandler(session_id)]})
except Exception as e:
    if "Kill switch activated" in str(e):
        # ユーザーへの通知
        return {
            "status": "terminated",
            "message": "管理者により処理が中断されました。",
            "partial_results": get_partial_results() # ここまでの成果物を返す
        }
    else:
        raise e

このように、エラーハンドリングを丁寧に行うことで、ガバナンス発動時のユーザー体験（UX）の低下を防ぎつつ、システムを安全に保護できます。さらに、中断時のコンテキストを保持しておくことで、後続のリカバリプロセスへの移行もスムーズに行えるようになります。

4. 実装ステップ2：トークン消費に基づく動的スロットリング

次に、コスト管理の本丸である「トークン予算」の実装です。実行回数ではなく、実際のトークン消費量に基づいて制御する高度なアプローチです。

TikToken等を用いたリアルタイムトークン計算の実装

OpenAIのモデルを使用する場合、tiktoken ライブラリを使って正確なトークン数を計算できます。これをコールバック処理などに組み込み、累積消費量をリアルタイムで計測します。

また、最新のエージェントフレームワーク（LangGraphなど）では、DynamoDBSaverのようなチェックポイント機能を用いた高度な状態の永続化が主流になりつつありますが、ここでは基本的なKVS（Redisなど）を用いた状態管理の例を示します。

import tiktoken

class TokenBudgetCallbackHandler(BaseCallbackHandler):
    def __init__(self, session_id: str, max_tokens: int, model_name: str = "gpt-4o"):
        self.session_id = session_id
        self.max_tokens = max_tokens
        self.encoding = tiktoken.encoding_for_model(model_name)
        self.current_usage = 0
        # Redisから現在の使用量を取得（LangGraphのDynamoDBSaver等に置き換えも可能）
        stored_usage = r.get(f"session:{session_id}:usage")
        if stored_usage:
            self.current_usage = int(stored_usage)

    def _check_budget(self):
        if self.current_usage >= self.max_tokens:
            raise Exception(f"Token budget exceeded: {self.current_usage} / {self.max_tokens}")

    def on_llm_start(self, serialized, prompts, kwargs):
        # 入力プロンプトのトークン見積もり
        prompt_tokens = sum(len(self.encoding.encode(p)) for p in prompts)
        
        # 実行前に予算チェック（入力分を加算して評価）
        if self.current_usage + prompt_tokens > self.max_tokens:
             raise Exception("Insufficient token budget for this request.")

    def on_llm_end(self, response, kwargs):
        # 実際の消費量（入力+出力）を正確に加算
        token_usage = response.llm_output.get("token_usage", {})
        total_tokens = token_usage.get("total_tokens", 0)
        
        self.current_usage += total_tokens
        
        # Redisへ更新（有効期限を設定してストレージを節約）
        r.set(f"session:{self.session_id}:usage", self.current_usage)
        r.expire(f"session:{self.session_id}:usage", 3600) # 1時間有効
        
        self._check_budget()

累積コストに基づく動的な実行停止ロジック

上記のコードのポイントは、on_llm_start（実行前）と on_llm_end（実行後）の両方でチェックを行っている点です。

  • 事前チェック: これから送信するプロンプトが巨大すぎて予算を超える場合、APIを呼び出す前に処理を停止し、無駄なコスト発生を防ぎます。
  • 事後チェック: 実際に生成された回答が長すぎて予算を超えた場合、エージェントを次のステップへ進ませないように制御します。

この「二重チェック」機構により、予算超過を最小限に抑えられます。チェックポイントAPIを活用する環境では、このタイミングでエージェントの実行状態を保存し、後から安全に再開する設計も有効です。

ステップごとの消費傾向分析と異常検知

さらに高度な実装として、ステップごとのトークン消費量の「傾き」を監視するアプローチがあります。

通常、エージェントのコンテキストは対話や推論の進行に伴って徐々に増加しますが、急激なスパイク（例：あるステップで突然10倍のトークンを消費した）が発生した場合は、誤って大量のテキストデータを読み込んだ、あるいは検索ツールが想定外の巨大なレスポンスを返した可能性があります。

# 異常検知ロジックのイメージ
previous_usage = get_previous_step_usage(session_id)
current_step_usage = total_tokens

# 前回の5倍以上の消費を検知した場合
if current_step_usage > previous_usage * 5:
    log_warning("Abnormal token spike detected.")
    # 必要に応じて管理者にアラートを通知、またはエージェントの実行を一時停止

このようなヒューリスティックなルールを条件分岐に組み込むことで、無限ループや予期せぬ大量課金といった事故を未然に防げます。LangGraphなどのエージェント構築フレームワークでは、こうした異常検知をノード間の遷移条件（Conditional Edges）として定義することで、より堅牢なワークフローを構築できます。

5. 実装ステップ3：セマンティック・ループ検知の自動化

4. 実装ステップ2：トークン消費に基づく動的スロットリング - Section Image

最後に、技術的に非常に興味深い「意味的なループ」の検知を取り上げます。エージェントが「エラーが発生しました」→「再試行します」→「エラーが発生しました」というやり取りを延々と繰り返すケースは珍しくありません。このような状況では、単純なトークン消費の監視だけでは異常に気づきにくいという課題があります。

埋め込みベクトルを用いた「同じ思考の繰り返し」検知

この問題を解決するためには、エージェントの思考（Thought）や行動（Action）の履歴をベクトル化し、過去の行動との類似度を計算するアプローチが有効です。直近の行動と過去の行動が酷似していれば、エージェントがスタックしているとシステム側で判断できます。

from langchain_openai import OpenAIEmbeddings
import numpy as np

# OpenAIの埋め込みモデルを利用
embeddings = OpenAIEmbeddings()

def calculate_similarity(text1, text2):
    vec1 = embeddings.embed_query(text1)
    vec2 = embeddings.embed_query(text2)
    return np.dot(vec1, vec2) # コサイン類似度（正規化前提）

class SemanticLoopDetector:
    def __init__(self, history_window=3, threshold=0.95):
        self.history = [] # 直近の思考履歴
        self.window = history_window
        self.threshold = threshold

    def check_loop(self, current_thought):
        for past_thought in self.history[-self.window:]:
            similarity = calculate_similarity(current_thought, past_thought)
            if similarity > self.threshold:
                return True # ループ検知
        
        self.history.append(current_thought)
        return False

コサイン類似度によるアクションの重複判定

作成した SemanticLoopDetector をエージェントの実行ループ内に組み込みます。

例えば、エージェントが「Pythonで特定のコードを実行する」というアクションを生成した際、そのコード内容や意図が前回の失敗時とほぼ同じ（類似度0.95以上）であれば、「そのアプローチは先ほど失敗しています」とシステム側から介入する必要があります。

最近の動向として、LangGraphのCheckpoints APIなどを活用した状態管理（永続化）の仕組みも進化を続けています。特に、DynamoDBSaverなどと統合したチェックポイント機能による状態追跡と、ベクトル検索を用いた類似度判定を組み合わせることで、より堅牢なループ検知システムを構築できます。ただし、AIエージェント構築フレームワークのアップデートは非常に速いため、実装の際は公式ドキュメントで最新のCheckpoints APIの仕様や推奨構成を確認することをお勧めします。

進捗がない場合の軌道修正プロンプトの自動注入

ループを検知した際、単に処理を強制終了させるのではなく、エージェントに「気づき」を与えるプロンプトを注入するのがスマートな解決策と言えます。

# ループ検知時の処理
if detector.check_loop(agent_action.log):
    # ツール実行をスキップし、システムメッセージを返す
    return {
        "tool_output": "System Alert: あなたは同じ行動を繰り返しています。アプローチを変更するか、タスクが不可能であると報告してください。"
    }

これにより、エージェントは自律的に状況を客観視し、別のアプローチ（検索ワードを変える、諦めるなど）へ方向転換できます。

GPT-4oやClaude 3.5 Sonnetのような高度なAPIモデルでは、推論機能によって自律的なタスク遂行能力が飛躍的に向上しています。しかし、未知のエラーに直面した際の無限ループのリスクは依然として残ります。旧世代のモデルから現行の高性能モデルへ移行する際にも、このようなAIエージェントの「メタ認知」を外部から補助するガバナンス機構を組み込んでおくことで、予期せぬ挙動を安全に制御し、システムの信頼性を高めることが可能です。

6. 本番運用とモニタリング体制

実装したガバナンス機能を運用に乗せるための体制について触れておきます。

コスト監視ダッシュボードの構築（Grafana/Datadog連携）

Redisに保存されたトークン消費データは、ログとして出力し、可視化ツールに取り込みましょう。

  • Total Token Usage per User: 誰がどれくらい使っているか。
  • Average Tokens per Task: 1タスクあたりの平均コスト。
  • Loop Limit Hit Rate: 何％のエージェントが強制停止されたか。

これらの指標は、ガバナンス設定（予算や回数制限）が適切かどうかを判断する材料になります。強制停止率が高すぎるなら、エージェントの能力不足か、制限が厳しすぎるかのどちらかです。

アラート設定のベストプラクティス

SlackやPagerDutyへの通知は、「異常」の定義を明確にしてから設定します。

  • Critical: キルスイッチ発動、予算超過による停止。
  • Warning: セマンティックループの検知、トークン消費の急激なスパイク。

すべての停止を通知するとオオカミ少年になるため、特に「想定外の挙動」に絞ってアラートを飛ばすのがコツです。

インシデント発生時の事後分析（Post-mortem）フロー

もし暴走が発生した場合、必ず「思考トレース（Thought Trace）」を分析してください。LangSmithなどのLLM開発用プラットフォームを使えば、どのプロンプトがトリガーとなってループに入ったかを詳細に追跡できます。

この分析結果を元に、システムプロンプト（System Prompt）に「〇〇のような状況では××してはいけない」という指示を追加し、モデル自体の挙動を改善していくPDCAサイクルを回しましょう。

まとめ

4. 実装ステップ2：トークン消費に基づく動的スロットリング - Section Image 3

AIエージェントのガバナンスは、単なるコスト削減策ではありません。それは、確率的な挙動をするAIを、確定的なビジネスプロセスの中に安全に統合するための「信頼性の基盤」です。

今回解説した3つのレイヤーを実装することで、以下のメリットが得られます。

  1. 予期せぬコスト超過の防止（動的トークンバジェット）
  2. 無限ループによるリソース枯渇の回避（セマンティック検知）
  3. 運用者の心理的安全性（キルスイッチと可視化）

技術的な準備は整いました。あとは、実際のシステムにこれらのコードを組み込むだけです。

自社のアーキテクチャに合わせた具体的なガバナンス設計のレビューや、より高度なLangGraphの実装にあたっては、専門家に相談することも一つの有効な手段です。PoCの壁を越え、安心してスケールできる本番環境の構築を目指しましょう。

AIエージェントの無限ループを防ぐ：トークン・ガバナンス自動化の実装パターン - Conclusion Image

コメント

コメントは1週間で消えます
0 / 150
コメントを読み込み中...