無限ループの検出と停止
最大反復回数のガード、反復アクションの検出、ループ遮断機構を学びます。
「無限ループの検出と停止」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
無限ループの脅威
無限ループに陥ったエージェントはトークンを消費し、リソースを占有する一方で、有用な出力を生成しません。本番環境では、これはそのまま無駄なコストとユーザーの不満につながります。
無限ループを防ぐには、最大反復回数の制限、繰り返されるアクションの検出、タイムアウトという3つの仕組みを組み合わせます。
ガード1:最大反復回数の制限
最もシンプルで重要なガードは、厳格なステップ数上限です。すべてのエージェントループに必ず設けてください。上限に達したら、エージェントは現時点での最善の回答、または明示的な失敗メッセージを返します。
MAX_ITERATIONS = 20
def run_agent(query: str) -> dict:
history = []
for step in range(1, MAX_ITERATIONS + 1):
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer'], 'steps': step}
result = execute_tool(action['tool'], action['args'])
history.append({'step': step, 'tool': action['tool'], 'result': result})
# Hard stop — max iterations reached
return {
'status': 'max_iterations_reached',
'answer': None,
'steps': MAX_ITERATIONS
}ガード2:繰り返されるアクションを検出する
繰り返されるアクションは、無限ループの典型的な特徴です。(tool_name、arguments)の組の履歴を追跡してください。同じ組がN回を超えて現れた場合、エージェントは行き詰まっています。ループを中断し、エラーメッセージを注入してください。
import hashlib
import json
def action_hash(tool_name: str, args: dict) -> str:
payload = json.dumps({'tool': tool_name, 'args': args}, sort_keys=True)
return hashlib.md5(payload.encode()).hexdigest()
def run_agent_with_repeat_detection(query: str) -> dict:
history = []
action_counts = {}
for step in range(1, 21):
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
key = action_hash(action['tool'], action['args'])
action_counts[key] = action_counts.get(key, 0) + 1
if action_counts[key] > 2: # seen this exact action more than twice
history.append({
'role': 'system',
'content': f'You have called {action["tool"]} with the same arguments {action_counts[key]} times. '
f'This approach is not working. Try a completely different strategy or state what you know so far.'
})
continue
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
return {'status': 'loop_detected', 'answer': None}ウィンドウ内で繰り返されるアクションを追跡する
全期間のカウントを追跡するのではなく、直近Nステップのスライディングウィンドウ内で繰り返しを検出します。これにより、わずかに変化しながら短い周期で循環するループを検出できます。
from collections import deque
def is_cycling(recent_actions: deque, window: int = 6) -> bool:
if len(recent_actions) < window:
return False
# Check if the last window/2 actions repeat the first window/2
half = window // 2
first_half = list(recent_actions)[:half]
second_half = list(recent_actions)[half:window]
return first_half == second_half
# In the agent loop:
# recent_actions = deque(maxlen=6)
# recent_actions.append(action_hash(tool, args))
# if is_cycling(recent_actions):
# print('Cycling detected — breaking loop')
# break
recent = deque(['a', 'b', 'a', 'b'], maxlen=6)
print(is_cycling(recent, window=4)) # True — cycling detectedsignal.alarmによる実時間タイムアウト
Unixシステムでは、signal.alarm()が指定した秒数の経過後にSIGALRMを発生させます。これにより、エージェントループが遅いツール呼び出しでブロックされている場合でも、強制的にタイムアウトできます。
import signal
class AgentTimeout(Exception):
pass
def timeout_handler(signum, frame):
raise AgentTimeout('Agent exceeded time limit')
def run_agent_with_signal_timeout(query: str, timeout_seconds: int = 60) -> dict:
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds) # set the alarm
try:
result = run_core_agent_loop(query)
signal.alarm(0) # cancel the alarm on success
return result
except AgentTimeout:
signal.alarm(0)
return {'status': 'timeout', 'answer': None}
except Exception as e:
signal.alarm(0)
raise
# Note: signal.alarm is Unix-only (Linux/Mac)threading.Timerによるタイムアウト(クロスプラットフォーム)
threading.TimerはWindowsを含むすべてのプラットフォームで動作します。タイムアウト時間の経過後にフラグを設定し、エージェントループでそのフラグを確認して、設定されていれば終了します。
import threading
def run_agent_with_timer_timeout(query: str, timeout_seconds: int = 60) -> dict:
timed_out = threading.Event()
def set_timeout():
timed_out.set()
timer = threading.Timer(timeout_seconds, set_timeout)
timer.start()
history = []
try:
for step in range(1, 21):
if timed_out.is_set():
return {'status': 'timeout', 'answer': None, 'steps': step}
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
finally:
timer.cancel() # always cancel if done before timeout
return {'status': 'max_steps', 'answer': None}asyncio.wait_forによる非同期タイムアウト
非同期エージェントのアーキテクチャでは、asyncio.wait_for(coroutine, timeout=N)を使用します。指定した秒数以内にコルーチンが完了しない場合、asyncio.TimeoutErrorが発生します。
import asyncio
async def run_async_agent(query: str) -> dict:
history = []
for step in range(1, 21):
action = await async_decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
result = await async_execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
return {'status': 'max_steps', 'answer': None}
async def run_with_timeout(query: str, timeout: float = 60.0) -> dict:
try:
return await asyncio.wait_for(run_async_agent(query), timeout=timeout)
except asyncio.TimeoutError:
return {'status': 'timeout', 'answer': None}
# asyncio.run(run_with_timeout('What is Python?', timeout=30.0))履歴に脱出指示を注入する
ループを検出したときは、何も記録せずに中断するだけにしないでください。何が起きたかを説明し、エージェントに結論を出すよう求めるシステムメッセージを会話履歴に注入します。これにより、処理を強制終了する前にLLMが最終回答を生成する機会を与えられます。
def inject_loop_escape(history: list, step: int, reason: str):
message = (
f'[SYSTEM] You have been running for {step} steps. Reason: {reason}. '
f'You MUST now provide a FINAL_ANSWER based on what you have found so far, '
f'even if the information is incomplete. Do not call any more tools.'
)
history.append({'role': 'system', 'content': message})
# In the agent loop, when approaching the limit:
# if step >= MAX_ITERATIONS - 2:
# inject_loop_escape(history, step, 'approaching max iteration limit')
# Or when a repeat is detected:
# if action_counts[key] > 2:
# inject_loop_escape(history, step, 'repeated action detected')
if __name__ == '__main__':
demo_history = []
inject_loop_escape(demo_history, step=18, reason='approaching max iteration limit')
print(demo_history[-1]['content'])
ループを中断したときにログを記録する
ループガードが発動したときは、必ずログに記録してください。これにより、エージェントが行き詰まる頻度と原因を記録できます。プロンプトやツール実装を改善するうえで、非常に価値のあるデータです。
import logging
logger = logging.getLogger('agent_guard')
def check_and_break_loop(step: int, action_counts: dict, current_key: str) -> bool:
count = action_counts.get(current_key, 0)
if count > 2:
logger.warning(
f'Infinite loop detected at step {step}. '
f'Action hash {current_key[:8]} seen {count} times. '
f'Breaking loop.'
)
return True # signal to break
if step >= 18: # approaching limit
logger.warning(
f'Approaching max iterations at step {step}. '
f'Injecting escape prompt.'
)
return False
if __name__ == '__main__':
import sys
logging.basicConfig(level=logging.WARNING, format='%(message)s', stream=sys.stdout)
demo_counts = {'search:{"q": "weather"}': 3}
check_and_break_loop(step=10, action_counts=demo_counts, current_key='search:{"q": "weather"}')
3つのガードを組み合わせる
本番環境のエージェントでは、3つのガードを組み合わせてください。最大反復回数(必須)、繰り返しの検出(循環を検出)、タイムアウト(ブロックするツール呼び出しを検出)です。これらを組み合わせることで、ループを堅牢に保護できます。
import threading
import hashlib
import json
def run_production_agent(query: str) -> dict:
MAX_STEPS = 20
TIMEOUT_SEC = 120
timed_out = threading.Event()
timer = threading.Timer(TIMEOUT_SEC, timed_out.set)
timer.start()
history = []
action_counts = {}
try:
for step in range(1, MAX_STEPS + 1):
if timed_out.is_set():
return {'status': 'timeout'}
action = decide_action(query, history)
if action['type'] == 'final_answer':
return {'status': 'ok', 'answer': action['answer']}
key = hashlib.md5(json.dumps(action, sort_keys=True).encode()).hexdigest()
action_counts[key] = action_counts.get(key, 0) + 1
if action_counts[key] > 2:
inject_loop_escape(history, step, 'repeat detected')
continue
result = execute_tool(action['tool'], action['args'])
history.append({'tool': action['tool'], 'result': result})
finally:
timer.cancel()
return {'status': 'max_steps'}単体テストでループガードをテストする
ループガード専用の単体テストを作成してください。同じツールを常に呼び出すモックエージェントを作り、想定したステップ数以内にガードがそれを検出することを確認します。
from unittest.mock import MagicMock
def test_repeat_detection_breaks_loop():
# Create a mock that always returns the same action
always_same_action = MagicMock(return_value={
'type': 'tool',
'tool': 'search_web',
'args': {'query': 'same query'}
})
always_success = MagicMock(return_value='some result')
result = run_agent_with_repeat_detection(
query='test',
decide_action=always_same_action,
execute_tool=always_success
)
# Should stop due to loop detection, not run all 20 steps
assert result['status'] in ('loop_detected', 'max_iterations_reached')
# Should not have run all 20 steps (loop should be detected by step 6-7)
print('Loop guard test passed')理解度チェック:無限ループの検出
無限ループを検出して中断する方法についての理解度を確認しましょう。
振り返り:無限ループの検出と中断
これで、3つの相補的なガードによってエージェントを無限ループから保護できるようになりました。
- 最大反復回数:すべてのエージェントループに必須の、厳格なステップ数上限
- 繰り返しの検出:(ツール、引数)の組をハッシュ化し、N回を超えて出現したら中断します
- タイムアウト:Unixでは
signal.alarm()、クロスプラットフォームではthreading.Timer、非同期処理ではasyncio.wait_for()を使用します - 脱出指示の注入:強制停止する前に、エージェントが結論を出す機会を与えます
- ガードが発動したときは必ずログに記録してください。プロンプト改善に役立つ貴重なデータになります
AI チューターと学ぶ AI Agents — 無料
ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。
- コース
- 60
- レッスン
- 239
よくある質問
「無限ループの検出と停止」レッスンは無料ですか?
はい。「無限ループの検出と停止」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「無限ループの検出と停止」で何を学びますか?
最大反復回数のガード、反復アクションの検出、ループ遮断機構を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか?
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「無限ループの検出と停止」レッスンにはどのくらい時間がかかりますか?
ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか?
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。