AIエージェントを実装していると、「目標を達成したはずなのにツール呼び出しを続ける」「同じ検索やAPI実行を何度も繰り返す」といったループ暴走に遭遇します。
私も手元で、LLMに「必要な情報を集めて答えを作る」という処理を任せた際、終了判定が曖昧なために同じステップを繰り返すケースを確認しました。原因は、LLMの性能だけではありません。エージェント側に、実行回数や時間、状態の変化を監視する仕組みが不足していました。
結論として、停止条件は1つに依存せず、次のような複数のガードを組み合わせるのが安全です。
- 最大ステップ数
- 最大実行時間
- 完了イベントの明示
- 同一状態の繰り返し検知
- ツール失敗回数の上限
- 外部からのキャンセル
この記事では、Pythonで最小限の実行ループを作り、どのように暴走を止めるかを整理します。
まず最大ステップ数を必ず設定する
最初に入れるべき制御は、最大ステップ数です。LLMが「まだ情報が足りない」と判断し続けても、指定回数を超えたら強制的に終了させます。
from dataclasses import dataclass, field
from typing import Any
@dataclass
class AgentState:
goal: str
steps: int = 0
history: list[dict[str, Any]] = field(default_factory=list)
MAX_STEPS = 8
def call_model(state: AgentState) -> dict[str, Any]:
"""
実際にはここでLLMを呼び出す。
actionは「tool」「finish」のいずれかを返す想定。
"""
if state.steps >= 3:
return {
"action": "finish",
"answer": "収集した情報をもとに回答を作成しました。"
}
return {
"action": "tool",
"tool": "search",
"input": {"query": state.goal}
}
def run_agent(goal: str) -> str:
state = AgentState(goal=goal)
while state.steps < MAX_STEPS:
decision = call_model(state)
state.steps += 1
state.history.append(decision)
if decision["action"] == "finish":
return decision["answer"]
if decision["action"] == "tool":
result = execute_tool(
decision["tool"],
decision["input"]
)
state.history.append({
"type": "tool_result",
"result": result
})
return "最大実行回数に到達したため、処理を終了しました。"
def execute_tool(name: str, arguments: dict[str, Any]) -> str:
return f"{name}の実行結果: {arguments}"
ここで重要なのは、MAX_STEPSをLLMの指示文だけで管理しないことです。プロンプトに「3回以内に終了してください」と書いても、最終的な強制力はありません。
回数制限は、必ずアプリケーション側のwhile条件やランナーに持たせます。また、stepsをどの単位で数えるかも決めておく必要があります。LLMの呼び出し単位で数えるのか、ツール呼び出しも別に数えるのかを曖昧にすると、想定より多く実行されます。
完了条件は自然言語ではなく構造化する
「十分な情報が集まったら終了」のような自然言語だけでは、終了判定が不安定になります。モデルの出力に、完了状態を表すフィールドを持たせます。
{
"status": "completed",
"answer": "回答本文",
"next_action": null
}
継続する場合は、次のようにします。
{
"status": "needs_more_info",
"answer": null,
"next_action": {
"tool": "search",
"arguments": {
"query": "追加で確認する内容"
}
}
}
Python側では、許可された状態だけを受け付けます。
ALLOWED_STATUS = {
"completed",
"needs_more_info",
"failed",
}
def validate_decision(decision: dict) -> None:
status = decision.get("status")
if status not in ALLOWED_STATUS:
raise ValueError(f"不正なstatusです: {status}")
if status == "completed" and not decision.get("answer"):
raise ValueError("completedなのにanswerがありません")
if status == "needs_more_info" and not decision.get("next_action"):
raise ValueError("継続状態なのにnext_actionがありません")
実際に試したところ、finishという文字列を本文から検索する方式より、JSONのstatusを検証する方式のほうが、回答文中に「処理を終了します」と書かれてしまうケースに強くなりました。
ただし、JSON形式でも完全ではありません。パースエラーや必須フィールド不足を想定し、検証に失敗した場合は無制限に再試行せず、リトライ回数を制限します。
同じ状態の繰り返しを検知する
最大ステップ数は最後の防波堤です。その前に、同じ行動を繰り返していることを検知できると、無駄なAPI呼び出しを減らせます。
import json
import hashlib
def state_fingerprint(decision: dict, history: list[dict]) -> str:
recent = history[-4:]
payload = {
"decision": decision,
"recent": recent,
}
raw = json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def is_repeated(
fingerprint: str,
fingerprints: list[str],
threshold: int = 2,
) -> bool:
return fingerprints.count(fingerprint) >= threshold
ループ内では、直近の判断と履歴からフィンガープリントを作ります。
fingerprints: list[str] = []
while state.steps < MAX_STEPS:
decision = call_model(state)
fingerprint = state_fingerprint(
decision,
state.history,
)
if is_repeated(fingerprint, fingerprints):
return "同じ処理が繰り返されたため、安全のため終了しました。"
fingerprints.append(fingerprint)
state.steps += 1
# ここでツール実行や完了判定を行う
単純な文字列比較だけでは、引数の順番やJSONの空白が違うだけで別の状態として扱われます。そのため、sort_keys=Trueなどで正規化してから比較します。
一方で、同じツールを複数回呼ぶこと自体が正しい処理もあります。例えば、ページネーションを使う検索では、検索条件が同じでもページ番号が変わります。何を「同じ状態」とみなすかは、ツール名だけでなく、引数や取得済みデータも含めて設計する必要があります。
時間・失敗回数・コストにも上限を置く
エージェントの実行ループと、複数の停止条件を組み合わせた流れは次の通りです。
実行回数が少なくても、1回のツール呼び出しが長時間ハングすると、エージェント全体が戻ってきません。そこで、時間制限と失敗回数も別に管理します。
import time
MAX_SECONDS = 30
MAX_TOOL_ERRORS = 3
def run_with_limits(goal: str) -> str:
started_at = time.monotonic()
tool_errors = 0
state = AgentState(goal=goal)
while state.steps < MAX_STEPS:
elapsed = time.monotonic() - started_at
if elapsed >= MAX_SECONDS:
return "実行時間の上限に到達したため終了しました。"
decision = call_model(state)
state.steps += 1
if decision["action"] == "finish":
return decision["answer"]
try:
result = execute_tool(
decision["tool"],
decision["input"],
)
except TimeoutError:
tool_errors += 1
if tool_errors >= MAX_TOOL_ERRORS:
return "ツールの失敗回数が上限に達しました。"
continue
except Exception:
tool_errors += 1
if tool_errors >= MAX_TOOL_ERRORS:
return "ツール実行に繰り返し失敗したため終了しました。"
continue
state.history.append({
"decision": decision,
"result": result,
})
return "最大実行回数に到達したため終了しました。"
ここでtime.time()ではなくtime.monotonic()を使っているのは、システム時刻の変更による影響を避けるためです。
また、本番運用では、実行回数だけでなくLLMのトークン数や外部APIの呼び出し数も記録します。予算を超えたら、処理を中断するか、要約モードへ切り替える設計にします。
ログで「なぜ止まったか」を残す
停止した事実だけでは、後から原因を追えません。最低限、次の情報をログに残すようにしました。
def log_stop_reason(
reason: str,
steps: int,
elapsed_seconds: float,
) -> None:
print({
"event": "agent_stopped",
"reason": reason,
"steps": steps,
"elapsed_seconds": round(elapsed_seconds, 3),
})
停止理由には、例えば次の値を使います。
completedmax_stepstimeoutrepeated_statetool_error_limitcancelledinvalid_model_output
手元でテストするときは、意図的に終了しないモックを作ると確認しやすくなります。
def endless_model(state: AgentState) -> dict:
return {
"action": "tool",
"tool": "search",
"input": {"query": "同じ検索"},
}
このモックを使い、最大ステップ数を超えた時点で必ず戻ること、ログの停止理由がmax_stepsになることを確認します。停止処理は「たぶん動く」ではなく、無限ループを再現したテストで検証するのがポイントです。
まとめ
AIエージェントのループ暴走は、モデルに終了をお願いするだけでは防げません。アプリケーション側で、実行を有限な処理として設計する必要があります。
特に最初に実装したいのは、最大ステップ数、最大実行時間、構造化された完了状態です。さらに、同一状態の繰り返し検知、ツール失敗回数、トークンやAPI利用量の上限を加えると、異常時にも制御しやすくなります。
私が実装時に意識しているのは、「正常終了できること」と同じくらい「正常終了できなくても安全に止まること」です。エージェントの賢さを高める前に、必ず停止できる仕組みを用意しておくと、デバッグもしやすく、運用時の予期しないコストや障害も抑えられます。