PricingLoginStart free
AI Agents · 课时

检测并打破无限循环

最大迭代次数保护、重复操作检测和循环断路器。

第 3 / 4 课13 个步骤

检测并打破无限循环 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

无限循环威胁

陷入无限循环的代理会消耗令牌、占用资源，并且无法产生有用的输出。在生产环境中，这会直接转化为金钱浪费和用户挫败感。

三种机制协同工作，可以防止无限循环：最大迭代次数限制、重复操作检测和超时机制。

防护机制 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：检测重复操作

重复操作是无限循环的典型特征。请跟踪（工具名称、参数）对的历史记录。如果同一对组合出现超过 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 detected

防护机制 3：使用 signal.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"}')

组合三种防护机制

生产环境中的代理应组合三种防护机制：最大迭代次数（必须设置）、重复检测（捕获循环）和超时（捕获阻塞的工具调用）。三者结合可以使循环坚不可摧。

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')

知识检查：检测无限循环

请测试您对无限循环检测和中断技术的理解。

在交互式课程中回答

回顾：检测并中断无限循环

您现在可以使用三种互补的防护机制来保护代理，避免陷入无限循环：

  • 最大迭代次数：硬性步骤上限，每个代理循环都必须设置
  • 重复检测：对（工具、参数）对进行哈希处理，在出现次数超过 N 次时中断
  • 超时：在 Unix 上使用 signal.alarm()，跨平台使用 threading.Timer，异步场景使用 asyncio.wait_for()
  • 退出指令注入：在强制停止之前，给代理一个得出结论的机会
  • 防护机制触发时务必记录日志——这些数据对改进提示词非常有价值
免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程
60
课程
239
免费开始学习查看完整课程

常见问题解答

「检测并打破无限循环」课时是免费的吗？

是的 — 「检测并打破无限循环」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「检测并打破无限循环」这节课中我会学到什么？

最大迭代次数保护、重复操作检测和循环断路器。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。

「检测并打破无限循环」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 常见的代理循环故障
  2. 代理步骤的跟踪日志
  3. 检测并打破无限循环
  4. 逐步调试技术
← 返回 AI Agents