引言 Agent系统最令人头疼的故障模式之一就是无限循环——Agent反复调用同一个工具、在两个状态间来回切换、或者陷入"思考但不行动"的死循环。这类问题不仅浪费Token和计算资源，还可能导致用户长时间等待无响应。
2026年，随着Agent自主能力的增强（如AutoGPT式的自主规划），循环检测和超时控制变得更加关键。一个能够自主决策的Agent，如果不能有效检测和打破循环，其危害性远大于传统软件的死循环。
循环类型分析 Agent系统中的四种典型循环 类型1：工具调用循环 类型2：状态转移循环 ┌─────────────────┐ ┌──────────────────┐ │ Agent ──▶ Tool A│ │ State A ──▶State B│ │ ▲ │ │ │ ▲ │ │ │ └──────┘ │ │ └─────────┘ │ │ (反复调用同一工具)│ │ (状态来回切换) │ └─────────────────┘ └──────────────────┘ 类型3：推理循环 类型4：工具链循环 ┌─────────────────┐ ┌──────────────────┐ │ Think ──▶ Think │ │ Tool A ──▶Tool B │ │ ▲ │ │ │ ▲ │ │ │ └─────────┘ │ │ └─────────┘ │ │ (反复思考不行动) │ │ (工具间互相触发) │ └─────────────────┘ └──────────────────┘ 循环检测算法 基于状态指纹的检测 import hashlib from collections import defaultdict from dataclasses import dataclass @dataclass class CycleDetector: """基于状态指纹的循环检测器""" max_history: int = 50 # 保留最近50步 cycle_threshold: int = 3 # 重复出现3次判定为循环 def __init__(self): self.state_history: list = [] self.fingerprint_counts: dict = defaultdict(int) def record_state(self, state: dict) -> bool: """记录状态，返回是否检测到循环""" # 生成状态指纹 fingerprint = self._generate_fingerprint(state) self.state_history.append({ "fingerprint": fingerprint, "state": state, "timestamp": datetime.now() }) # 限制历史长度 if len(self.state_history) > self.max_history: old = self.state_history.pop(0) self.fingerprint_counts[old["fingerprint"]] -= 1 self.fingerprint_counts[fingerprint] += 1 # 检测循环 if self.fingerprint_counts[fingerprint] >= self.cycle_threshold: return True # 检测模式循环（A-B-A-B模式） if self._detect_pattern_cycle(): return True return False def _generate_fingerprint(self, state: dict) -> str: """生成状态指纹""" # 提取关键状态信息 key_info = { "intent": state.get("intent"), "active_tool": state.get("active_tool"), "tool_params_hash": hashlib.md5( json.dumps(state.get("tool_params", {}), sort_keys=True).encode() ).hexdigest()[:8], "fsm_state": state.get("fsm_state"), "pending_actions": state.get("pending_actions", []) } fingerprint_str = json.dumps(key_info, sort_keys=True) return hashlib.md5(fingerprint_str.encode()).hexdigest() def _detect_pattern_cycle(self) -> bool: """检测模式循环（如A-B-A-B）""" if len(self.state_history) < 4: return False # 检查最近的状态是否形成周期模式 recent = [s["fingerprint"] for s in self.state_history[-6:]] # 尝试不同的周期长度 for period in range(2, 4): if len(recent) >= period * 2: pattern = recent[-period:] previous = recent[-period*2:-period] if pattern == previous: return True return False 基于行为序列的检测 class BehaviorSequenceAnalyzer: """基于行为序列的循环检测""" def __init__(self): self.action_sequences = [] def analyze(self, actions: list) -> dict: """分析行为序列""" result = { "has_cycle": False, "cycle_type": None, "cycle_length": 0, "suggestion": None } # 使用Floyd算法检测环 cycle = self._floyd_cycle_detection(actions) if cycle: result["has_cycle"] = True result["cycle_length"] = len(cycle) result["cycle_type"] = self._classify_cycle(cycle) result["suggestion"] = self._suggest_break_strategy(cycle) return result def _floyd_cycle_detection(self, sequence: list) -> list: """Floyd环检测算法""" if len(sequence) < 2: return None # 快慢指针 slow = 0 fast = 0 while True: slow = (slow + 1) % len(sequence) fast = (fast + 2) % len(sequence) if sequence[slow] == sequence[fast]: # 找到环，确定环的起始和长度 start = 0 ptr1 = start ptr2 = slow while ptr1 != ptr2: ptr1 = (ptr1 + 1) % len(sequence) ptr2 = (ptr2 + 1) % len(sequence) # 提取环 cycle_start = ptr1 cycle = [sequence[cycle_start]] ptr = (cycle_start + 1) % len(sequence) while ptr != cycle_start: cycle.append(sequence[ptr]) ptr = (ptr + 1) % len(sequence) return cycle if fast == 0: return None # 无环 def _classify_cycle(self, cycle: list) -> str: """分类循环类型""" tools_in_cycle = [a for a in cycle if a.get("type") == "tool_call"] thoughts_in_cycle = [a for a in cycle if a.get("type") == "thought"] if len(tools_in_cycle) == len(cycle): return "tool_cycle" elif len(thoughts_in_cycle) == len(cycle): return "reasoning_cycle" else: return "mixed_cycle" def _suggest_break_strategy(self, cycle: list) -> str: """建议打破循环的策略""" cycle_type = self._classify_cycle(cycle) strategies = { "tool_cycle": "尝试用不同参数调用工具，或切换到替代工具", "reasoning_cycle": "强制进入执行阶段，或向用户请求澄清", "mixed_cycle": "重置上下文窗口，或回退到上一个成功状态" } return strategies.get(cycle_type, "终止当前任务并重试") 多级超时控制 class MultiLevelTimeout: """多级超时控制体系""" LEVELS = { "tool_call": { "default": 30, # 单次工具调用 "search": 15, # 搜索类工具 "code_exec": 60, # 代码执行 "file_io": 10, # 文件操作 }, "step": { "understanding": 10, # 意图理解 "planning": 15, # 规划 "retrieving": 10, # 检索 "executing": 120, # 工具执行 "generating": 30, # 响应生成 }, "session": { "max_duration": 600, # 单次会话最大10分钟 "idle_timeout": 120, # 空闲2分钟超时 }, "workflow": { "max_steps": 50, # 工作流最大步骤数 "max_tool_calls": 20, # 最大工具调用次数 "max_tokens": 100000, # 最大Token消耗 } } def __init__(self): self.active_timers = {} async def with_timeout( self, level: str, operation: str, coro: asyncio.coroutines ): """带超时执行协程""" timeout = self.LEVELS.get(level, {}).get(operation, 30) try: result = await asyncio.wait_for(coro, timeout=timeout) return result except asyncio.TimeoutError: logger.warning( f"Timeout at {level}.{operation} after {timeout}s" ) await self._handle_timeout(level, operation) raise async def _handle_timeout(self, level: str, operation: str): """超时处理策略""" if level == "tool_call": # 记录工具超时，可能降级 await self._record_tool_timeout(operation) elif level == "step": # 步骤超时，尝试跳过或降级 await self._try_degrade_step(operation) elif level == "session": # 会话超时，优雅终止 await self._graceful_session_end() elif level == "workflow": # 工作流超限，强制终止 await self._force_terminate() 循环打破与自恢复 class CycleBreaker: """循环打破器""" def __init__(self, cycle_detector, timeout_controller): self.detector = cycle_detector self.timeout = timeout_controller async def monitor_and_break( self, agent_session, check_interval: float = 1.0 ): """持续监控并打破循环""" while not agent_session.is_complete(): current_state = agent_session.get_state() if self.detector.record_state(current_state): logger.warning("Cycle detected, initiating break sequence") await self._break_cycle(agent_session) return True await asyncio.sleep(check_interval) return False async def _break_cycle(self, agent_session): """执行循环打破策略""" # 策略1：注入扰动——修改Agent的上下文 perturbation = { "system_message": ( "你似乎陷入了循环。请尝试不同的方法，" "或者明确说明你无法完成此任务。" ), "temperature_boost": 0.3, # 提高温度增加随机性 "disable_repeated_tool": True # 禁用循环工具 } await agent_session.inject_perturbation(perturbation) # 策略2：重置到上一个健康状态 healthy_state = agent_session.get_last_healthy_state() if healthy_state: await agent_session.restore_state(healthy_state) # 策略3：降级为简化流程 await agent_session.switch_to_degraded_mode( simplified_tools=True, max_steps=5 ) # 策略4：最终兜底——请求人工介入 if not await agent_session.try_recover(): await agent_session.escalate_to_human( reason="无法自动打破循环", context=agent_session.get_debug_context() ) 生产实践：超时配置矩阵 场景 工具超时 步骤超时 会话超时 最大步数 最大Token 简单问答 10s 15s 60s 5 5K 工具调用 30s 120s 300s 15 30K 复杂分析 60s 300s 600s 30 80K 自主任务 120s 600s 1800s 50 200K 批处理 300s 1800s 7200s 100 500K 监控与告警 class CycleMonitor: """循环监控器""" async def collect_metrics(self) -> dict: return { "cycle_detected_rate": await self._get_rate("cycle_detected"), "timeout_rate_by_level": { "tool": await self._get_rate("timeout.tool_call"), "step": await self._get_rate("timeout.step"), "session": await self._get_rate("timeout.session"), }, "avg_steps_to_cycle": await self._get_avg("steps_before_cycle"), "break_success_rate": await self._get_rate("cycle_break_success"), "human_escalation_rate": await self._get_rate("human_escalation"), } 总结 循环检测和超时控制是Agent系统鲁棒性的基石。基于状态指纹的检测能够高效识别精确循环，基于行为序列的分析能够发现模式循环。多级超时体系确保任何级别的异常都有对应的兜底机制。当检测到循环时，系统应按照"注入扰动→重置状态→降级模式→人工介入"的顺序尝试恢复。
...
