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

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"),
        }