AI Agent防无限循环指南：多层断路器+负载哈希+财务切断机制

生产环境中的瓶颈

运行在工具调用循环中的自主 AI Agent 表现不可预测。当 LLM 遇到意外的 schema、瞬态网络错误或歧义的 prompt 时，往往会进入一种幻觉重试风暴。

在标准 Web 应用中，无限循环会触发速率限制或返回 500 Internal Server Error。而在 Agentic 架构中，不受约束的 ReAct 循环会持续执行外部 API 调用，消耗 Token、耗尽上游配额，并在几分钟内产生巨额云服务账单。

以下是这个在大量代码库中运行的反模式：

# Anti-pattern: Unbounded autonomous agent loop
while not task_complete:
    action = llm.decide_action(state)
    result = external_api.call(action.endpoint, action.params)
    state = update_state(result)

如果 LLM 因无法解析的响应而无法转换状态，此循环将无限期运行。云服务提供商不会为自残式的 API 调用退款。

系统架构与解决方案

要使 AI Agent 工具执行达到生产级别的安全性，绝不允许从 Agent 代码直接调用 API。应当将每个外部请求路由通过一个隔离的 API Safety Wrapper，并实现三层独立的防御机制：

确定性请求防火墙：每个任务会话的执行次数硬性上限（TTL 计数器）。

滑动窗口循环检测器：对传出请求 Payload 进行哈希，以捕获重复或振荡的工具调用。

财务熔断器：预检预算验证器，当预期成本超出会话限制时立即切断凭证。

[ AI Agent Engine ]
        │
        ▼
[ API Safety Wrapper ]
   ├── 1. Call Counter Check (Limit < N)
   ├── 2. Hash Duplicate Detector (Window: last 3 calls)
   └── 3. Pre-flight Cost Estimator (Budget < Limit)
        │
   ┌────┴──────────────────────────┐
[ Passed ]                    [ Tripped ]
   │                               │
   ▼                               ▼
[ External Upstream API ]     [ Emergency Kill Switch ]
                              (Revoke Token & Abort)

这样可以确保即使 Agent 产生幻觉或崩溃，其影响范围也被严格限制在单个会话预算内。

以下是一个轻量级的、生产级别的安全包装器，可以包裹任何 HTTP 客户端或 SDK：

import hashlib

class APISafetyWrapper:
    def __init__(self, client, max_calls: int = 50, budget_limit: float = 5.0):
        self.client = client
        self.max_calls = max_calls
        self.budget_limit = budget_limit
        self.history = []
        self.total_cost = 0.0

    def execute(self, endpoint: str, payload: dict, estimated_cost: float = 0.02):
        sig = hashlib.md5(f"{endpoint}:{sorted(payload.items())}".encode()).hexdigest()

        if len(self.history) >= self.max_calls:
            raise RuntimeError(f"Circuit Breaker: Hard limit ({self.max_calls}) reached.")

        if self.history[-3:].count(sig) >= 2:
            raise RuntimeError(f"Loop Detected: Repeating payload sent to {endpoint}.")

        if (self.total_cost + estimated_cost) > self.budget_limit:
            self.client.revoke_credentials()  # Emergency shutdown
            raise PermissionError("Budget Exceeded: Financial kill-switch triggered.")

        self.history.append(sig)
        self.total_cost += estimated_cost
        return self.client.call(endpoint, payload)