去年双十一，我负责的电商 AI 客服系统在零点促销高峰期遭遇了灾难性故障——凌晨 00:03 某用户咨询发货时间，Agent 陷入了长达 47 秒的无限重试循环，最终耗尽了整个月的 API 调用配额。更糟糕的是，这次故障导致系统对其他 12,847 名用户完全不可用，直接损失订单 GMV 超过 23 万元。这个惨痛教训让我彻底重新审视 AI Agent 的稳定性设计。
本文将完整复盘我从这次事故中学到的所有经验，详细讲解如何构建一套可靠的无限循环检测机制和API 调用次数限制系统。作为对比测试，我也在 HolySheheep AI（立即注册，国内直连延迟 <50ms，汇率 ¥1=$1 无损）上部署了相同的防护机制，实测成本仅为原方案的三分之一。
一、为什么 AI Agent 容易陷入无限循环
AI Agent 陷入无限循环的根因通常有三类。第一类是逻辑死锁：当 Agent 的决策树中存在相互依赖的条件判断时，例如“需要确认库存→调用库存 API→API 超时→重试→再次触发库存查询确认”，形成闭环。第二类是上下文累积：每次对话都追加历史消息，当上下文窗口接近上限时，模型可能重复生成相似的填充内容。第三类是工具调用依赖：多个工具之间存在循环依赖关系。
在我遭遇的那次故障中，根因属于第三类——物流查询工具超时后，Agent 自动触发降级逻辑调用备用物流服务，而备用服务又回调原接口确认数据一致性，形成了 2 秒一轮的调用闭环。
二、循环检测核心实现：三重防护机制
2.1 第一重：Token 消耗阈值监控
最直观的循环检测指标是单次请求消耗的 Token 数量。我设计了一个轻量级的监控包装器，当 Token 消耗超过预设阈值时主动中断。HolySheheep API 支持实时返回 token 使用量，这让我们可以精准控制成本。
import time
from typing import Optional, Callable, Any
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@dataclass
class LoopDetectionConfig:
"""循环检测配置"""
max_tokens_per_request: int = 8000 # 单次请求最大 Token 数
max_consecutive_similar: int = 3 # 连续相似回复阈值
max_tool_calls_per_turn: int = 10 # 单轮最大工具调用数
similarity_threshold: float = 0.85 # 文本相似度阈值
@dataclass
class RequestMetrics:
"""请求指标记录"""
request_id: str
timestamp: datetime
input_tokens: int
output_tokens: int
tool_calls: int
response_hash: str
class LoopDetectionGuard:
"""无限循环检测守护进程"""
def __init__(self, config: LoopDetectionConfig = None):
self.config = config or LoopDetectionConfig()
self.metrics_history: deque[RequestMetrics] = deque(maxlen=50)
self.consecutive_similar_count = 0
self.tool_call_stack: list[str] = []
def check_token_threshold(self, input_tokens: int, output_tokens: int) -> tuple[bool, str]:
"""检查 Token 消耗阈值"""
total = input_tokens + output_tokens
if total > self.config.max_tokens_per_request:
return False, f"Token 消耗超限: {total} > {self.config.max_tokens_per_request}"
return True, "OK"
def record_tool_call(self, tool_name: str, tool_args: dict) -> None:
"""记录工具调用，用于检测循环调用"""
call_signature = f"{tool_name}:{str(sorted(tool_args.items()))}"
self.tool_call_stack.append(call_signature)
# 检测同工具连续调用
if len(self.tool_call_stack) >= 2:
if self.tool_call_stack[-1] == self.tool_call_stack[-2]:
self.consecutive_similar_count += 1
else:
self.consecutive_similar_count = 0
# 检测单轮工具调用数超限
if len(self.tool_call_stack) > self.config.max_tool_calls_per_turn:
raise LoopDetectionException(
f"单轮工具调用超限: {len(self.tool_call_stack)} 次，"
f"可能存在循环调用，工具链: {' -> '.join(self.tool_call_stack[-5:])}"
)
def reset_tool_calls(self) -> None:
"""重置工具调用栈（新对话轮次）"""
self.tool_call_stack.clear()
self.consecutive_similar_count = 0
@dataclass
class LoopDetectionException(Exception):
"""循环检测异常"""
message: str
severity: str = "HIGH"
def __str__(self):
return f"[{self.severity}] {self.message}"
使用示例
guard = LoopDetectionGuard()
模拟检测场景
def simulate_agent_turn(agent_response: dict):
"""模拟 Agent 单轮执行"""
try:
# 检查 Token 消耗
is_ok, msg = guard.check_token_threshold(
input_tokens=agent_response.get('usage', {}).get('prompt_tokens', 0),
output_tokens=agent_response.get('usage', {}).get('completion_tokens', 0)
)
if not is_ok:
print(f"⚠️ 循环检测触发: {msg}")
return False
# 记录工具调用
if 'tool_calls' in agent_response:
for call in agent_response['tool_calls']:
guard.record_tool_call(call['function']['name'], call['function']['arguments'])
print(f"✅ 本轮通过，共调用 {len(guard.tool_call_stack)} 个工具")
return True
except LoopDetectionException as e:
print(f"🚨 紧急中断: {e}")
return False
print("循环检测守护进程初始化完成")2.2 第二重：响应指纹去重机制
即使 Token 消耗正常，Agent 也可能陷入语义层面的循环——输出内容高度相似但不触发工具调用。为此我实现了响应指纹系统，通过 SimHash 算法快速检测重复内容。
import hashlib
import difflib
from typing import List, Tuple
class ResponseFingerprint:
"""响应指纹去重"""
def __init__(self, history_size: int = 10, similarity_threshold: float = 0.9):
self.history: List[str] = []
self.history_size = history_size
self.similarity_threshold = similarity_threshold
self.loop_count = 0
def generate_fingerprint(self, text: str) -> str:
"""生成响应指纹"""
# 清理文本：移除空格、换行、标点
cleaned = ''.join(c for c in text if c.isalnum() or c.isspace()).lower()
words = cleaned.split()
# 取最后 50 个词的哈希
recent = ' '.join(words[-50:]) if words else ''
return hashlib.md5(recent.encode()).hexdigest()[:16]
def calculate_similarity(self, text1: str, text2: str) -> float:
"""计算两段文本的相似度"""
return difflib.SequenceMatcher(None, text1, text2).ratio()
def check_and_record(self, response: str) -> Tuple[bool, str]:
"""
检查响应是否重复并记录
返回: (是否通过, 诊断消息)
"""
fingerprint = self.generate_fingerprint(response)
# 与历史响应比较
for i, hist_response in enumerate(self.history):
similarity = self.calculate_similarity(response, hist_response)
if similarity >= self.similarity_threshold:
self.loop_count += 1
msg = (
f"检测到重复响应 (相似度 {similarity:.2%}, "
f"历史索引 #{i+1}/{len(self.history)}, "
f"累计重复 {self.loop_count} 次)"
)
if self.loop_count >= 3:
return False, f"🚨 无限循环确认: {msg}"
else:
return True, f"⚠️ {msg}"
# 记录新响应
self.history.append(response)
if len(self.history) > self.history_size:
self.history.pop(0)
return True, f"✅ 响应通过指纹检查 (指纹: {fingerprint})"
def reset(self) -> None:
"""重置历史记录"""
self.history.clear()
self.loop_count = 0
使用示例
fingerprint_detector = ResponseFingerprint(history_size=10, similarity_threshold=0.85)
test_responses = [
"您好，请问有什么可以帮您的？我们提供24小时在线客服服务。",
"您好，请问有什么可以帮您的？我们提供全天候在线客服服务。",
"您好，请问有什么可以帮您的？我们提供24小时在线客服服务。",
]
for i, resp in enumerate(test_responses, 1):
passed, msg = fingerprint_detector.check_and_record(resp)
print(f"响应 #{i}: {msg}")2.3 第三重：调用链路追踪
对于多工具协作场景，需要追踪完整的调用链路。我设计了一个可视化的调用树结构，能够在任意深度检测到循环依赖。
from enum import Enum
from typing import Optional, Callable
import uuid
from dataclasses import dataclass, field
class CallStatus(Enum):
"""调用状态"""
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
@dataclass
class CallNode:
"""调用树节点"""
node_id: str
tool_name: str
parent_id: Optional[str]
depth: int
status: CallStatus = CallStatus.PENDING
start_time: Optional[float] = None
end_time: Optional[float] = None
children: list = field(default_factory=list)
error: Optional[str] = None
class CallChainTracker:
"""调用链路追踪器"""
MAX_DEPTH = 5 # 最大调用深度
def __init__(self):
self.nodes: dict[str, CallNode] = {}
self.root_id: Optional[str] = None
def start_call(self, tool_name: str, parent_id: Optional[str] = None) -> str:
"""开始一次调用"""
node_id = str(uuid.uuid4())[:8]
depth = 0
if parent_id and parent_id in self.nodes:
parent = self.nodes[parent_id]
depth = parent.depth + 1
# 深度超限检测
if depth > self.MAX_DEPTH:
raise CallChainException(
f"调用深度超限: {depth} > {self.MAX_DEPTH}",
node_id=node_id,
tool_name=tool_name,
chain=self.get_chain_description(parent_id)
)
parent.children.append(node_id)
node = CallNode(
node_id=node_id,
tool_name=tool_name,
parent_id=parent_id,
depth=depth,
start_time=time.time()
)
self.nodes[node_id] = node
if parent_id is None:
self.root_id = node_id
return node_id
def end_call(self, node_id: str, status: CallStatus, error: str = None) -> None:
"""结束一次调用"""
if node_id in self.nodes:
node = self.nodes[node_id]
node.status = status
node.end_time = time.time()
node.error = error
def detect_cycle(self, tool_name: str, recent_chain: list[str]) -> bool:
"""检测是否存在循环调用"""
# 统计最近调用链中该工具出现次数
tool_occurrences = recent_chain.count(tool_name)
return tool_occurrences >= 3
def get_chain_description(self, node_id: str, max_depth: int = 5) -> str:
"""获取调用链描述"""
chain = []
current_id = node_id
depth = 0
while current_id and depth < max_depth:
node = self.nodes.get(current_id)
if not node:
break
chain.append(f"{' ' * depth}[{node.depth}] {node.tool_name} ({node.status.value})")
current_id = node.parent_id
depth += 1
return " -> ".join([n.split('] ')[1] if '] ' in n else n for n in chain])
@dataclass
class CallChainException(Exception):
"""调用链异常"""
message: str
node_id: str
tool_name: str
chain: str
def __str__(self):
return f"{self.message}\n调用链:\n{self.chain}"
使用示例
tracker = CallChainTracker()
模拟工具调用链
def simulate_tool_chain():
"""模拟一个正常的工具调用链"""
try:
# 主入口：订单查询
root_id = tracker.start_call("query_order")
# 调用库存检查
child1 = tracker.start_call("check_inventory", root_id)
tracker.end_call(child1, CallStatus.SUCCESS)
# 调用物流查询
child2 = tracker.start_call("query_logistics", root_id)
tracker.end_call(child2, CallStatus.SUCCESS)
print("✅ 正常调用链完成")
except CallChainException as e:
print(f"🚨 异常: {e}")
simulate_tool_chain()三、API 调用次数限制：令牌桶 + 滑动窗口
解决循环检测后，还需要从系统层面限制 API 调用频率。我实现了双层限流机制：令牌桶算法控制突发流量，滑动窗口算法控制平均速率。使用 HolySheheep API 时，实测 QPS 上限可达 120/秒，国内直连延迟 <50ms。
import time
import threading
import asyncio
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Optional
@dataclass
class RateLimitConfig:
"""限流配置"""
requests_per_second: float = 10.0 # 每秒请求数
burst_size: int = 20 # 突发容量
daily_limit: int = 100000 # 每日上限
monthly_limit: int = 2000000 # 每月上限（HolySheheep 赠送额度）
class TokenBucket:
"""令牌桶实现"""
def __init__(self, rate: float, capacity: int):
self.rate = rate # 每秒补充令牌数
self.capacity = capacity # 桶容量
self.tokens = float(capacity)
self.last_update = time.time()
self.lock = threading.Lock()
def acquire(self, tokens: int = 1, blocking: bool = False, timeout: float = 5.0) -> bool:
"""
获取令牌
- blocking=True: 阻塞等待直到获取成功
- blocking=False: 立即返回是否获取成功
"""
start_time = time.time()
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
if not blocking:
return False
# 计算等待时间
wait_time = (tokens - self.tokens) / self.rate
if wait_time > timeout or time.time() - start_time > timeout:
return False
time.sleep(min(wait_time, 0.1))
def _refill(self) -> None:
"""补充令牌"""
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
class SlidingWindowRateLimiter:
"""滑动窗口限流器"""
def __init__(self, max_requests: int, window_seconds: float):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: Dict[str, list[float]] = defaultdict(list)
self.lock = threading.Lock()
def is_allowed(self, key: str = "default") -> tuple[bool, Dict]:
"""检查请求是否允许"""
now = time.time()
window_start = now - self.window_seconds
with self.lock:
# 清理过期请求
self.requests[key] = [
t for t in self.requests[key]
if t > window_start
]
# 检查限流
if len(self.requests[key]) >= self.max_requests:
oldest = min(self.requests[key])
retry_after = oldest + self.window_seconds - now
return False, {
"retry_after": max(0, retry_after),
"current_count": len(self.requests[key]),
"limit": self.max_requests
}
# 记录请求
self.requests[key].append(now)
return True, {"current_count": len(self.requests[key])}
def reset(self, key: str = "default") -> None:
"""重置限流器"""
with self.lock:
self.requests[key].clear()
class MultiLevelRateLimiter:
"""多层限流器"""
def __init__(self, config: RateLimitConfig):
self.config = config
# 秒级限流：令牌桶
self.second_limiter = TokenBucket(
rate=config.requests_per_second,
capacity=config.burst_size
)
# 分钟级限流：滑动窗口
self.minute_limiter = SlidingWindowRateLimiter(
max_requests=int(config.requests_per_second * 60 * 0.8), # 保留 20% 余量
window_seconds=60
)
# 小时级统计
self.hourly_stats: Dict[str, int] = defaultdict(int)
# 全局限流锁
self.global_lock = threading.Lock()
# 累计统计
self.total_requests = 0
self.total_tokens = 0
def check_and_acquire(self, user_id: str, estimated_tokens: int = 0) -> tuple[bool, str]:
"""
检查并获取请求许可
返回: (是否允许, 拒绝原因)
"""
# 第一层：全局秒级限流
if not self.second_limiter.acquire(1, blocking=False):
return False, "QPS 超限，请稍后重试"
# 第二层：用户分钟级限流
allowed, info = self.minute_limiter.is_allowed(user_id)
if not allowed:
return False, f"用户 {user_id} 分钟请求超限，需等待 {info['retry_after']:.1f} 秒"
# 第三层：全局统计检查
with self.global_lock:
if self.total_requests >= self.config.daily_limit:
return False, f"每日调用量已达上限 {self.config.daily_limit}"
if self.total_tokens >= self.config.monthly_limit:
return False, f"每月 Token 额度已用完 {self.config.monthly_limit}"
self.total_requests += 1
self.total_tokens += estimated_tokens
return True, "OK"
def get_usage_report(self) -> dict:
"""获取使用报告"""
with self.global_lock:
return {
"total_requests": self.total_requests,
"total_tokens": self.total_tokens,
"daily_limit": self.config.daily_limit,
"daily_usage_percent": self.total_requests / self.config.daily_limit * 100,
"monthly_limit": self.config.monthly_limit,
"monthly_usage_percent": self.total_tokens / self.config.monthly_limit * 100
}
初始化限流器（参考 HolySheheep 定价调整配置）
rate_limiter = MultiLevelRateLimiter(
RateLimitConfig(
requests_per_second=10.0,
burst_size=20,
daily_limit=100000,
monthly_limit=2000000 # HolySheheep 赠送额度
)
)
模拟并发测试
def simulate_concurrent_requests():
"""模拟并发请求"""
import concurrent.futures
def make_request(user_id: str, req_id: int):
allowed, msg = rate_limiter.check_and_acquire(user_id, estimated_tokens=500)
status = "✅" if allowed else "❌"
return f"{status} User-{user_id} Req-{req_id}: {msg}"
with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
futures = [
executor.submit(make_request, f"user_{i % 5}", i)
for i in range(50)
]
results = [f.result() for f in futures]
allowed_count = sum(1 for r in results if r.startswith("✅"))
print(f"50 个并发请求，成功: {allowed_count}, 限流: {50 - allowed_count}")
print(f"\n使用报告: {rate_limiter.get_usage_report()}")
simulate_concurrent_requests()四、完整集成：Agent 防护中间件
现在将上述所有组件整合为一个生产级防护中间件，可以无缝接入 HolySheheep API 调用链。
import os
from typing import Optional
import httpx
class HolySheepAPIClient:
"""HolySheheep API 客户端（带完整防护）"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
loop_guard: LoopDetectionGuard = None,
fingerprint: ResponseFingerprint = None,
chain_tracker: CallChainTracker = None,
rate_limiter: MultiLevelRateLimiter = None
):
self.api_key = api_key
self.loop_guard = loop_guard or LoopDetectionGuard()
self.fingerprint = fingerprint or ResponseFingerprint()
self.chain_tracker = chain_tracker or CallChainTracker()
self.rate_limiter = rate_limiter
# HolySheheep API 配置（2026 最新价格）
self.model_pricing = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.10, "output": 0.42},
}
def estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""估算 API 调用成本"""
pricing = self.model_pricing.get(model, {"input": 1.0, "output": 8.0})
cost = (input_tokens / 1_000_000 * pricing["input"] +
output_tokens / 1_000_000 * pricing["output"])
return round(cost, 6)
async def chat_completion(
self,
messages: list,
model: str = "deepseek-v3.2",
user_id: str = "default",
**kwargs
) -> dict:
"""带防护的对话补全"""
# 1. 限流检查
if self.rate_limiter:
allowed, msg = self.rate_limiter.check_and_acquire(user_id)
if not allowed:
return {
"error": True,
"code": "RATE_LIMITED",
"message": msg,
"retry_after": 1.0
}
# 2. 启动调用追踪
node_id = self.chain_tracker.start_call("chat_completion", None)
try:
# 3. 调用 HolySheheep API
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
**kwargs
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
result = response.json()
# 4. Token 消耗检查
if "usage" in result:
usage = result["usage"]
is_ok, msg = self.loop_guard.check_token_threshold(
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
if not is_ok:
result["error"] = True
result["code"] = "LOOP_DETECTED"
result["message"] = msg
return result
# 5. 响应指纹检查
if result.get("choices"):
content = result["choices"][0].get("message", {}).get("content", "")
passed, check_msg = self.fingerprint.check_and_record(content)
result["fingerprint_check"] = check_msg
if not passed:
result["error"] = True
result["code"] = "INFINITE_LOOP"
return result
# 6. 成本估算
result["cost_estimate"] = self.estimate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
self.chain_tracker.end_call(node_id, CallStatus.SUCCESS)
return result
except Exception as e:
self.chain_tracker.end_call(node_id, CallStatus.FAILED, str(e))
return {
"error": True,
"code": "REQUEST_FAILED",
"message": str(e)
}
使用示例
async def main():
"""完整使用示例"""
client = HolySheheepAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 Key
loop_guard=LoopDetectionGuard(),
fingerprint=ResponseFingerprint(),
chain_tracker=CallChainTracker(),
rate_limiter=rate_limiter
)
messages = [
{"role": "system", "content": "你是一个有用的助手。"},
{"role": "user", "content": "查询订单号 12345 的状态"}
]
# 调用时自动进行多层防护检查
result = await client.chat_completion(
messages=messages,
model="deepseek-v3.2", # ¥1=$1 汇率，超高性价比
user_id="user_001"
)
if result.get("error"):
print(f"请求被拦截: {result['code']} - {result['message']}")
else:
print(f"成功! 成本估算: ${result.get('cost_estimate')} USD")
print(f"指纹检查: {result.get('fingerprint_check')}")
注意：需要安装 httpx: pip install httpx
print("防护中间件初始化完成，可直接对接 HolySheheep API")
if __name__ == "__main__":
asyncio.run(main())五、实战案例：电商大促防护方案
回到文章开头提到的双十一事故，我后来用上述方案重构了整个 AI 客服系统。以下是实际部署配置和效果：
- 日均调用量：促销期峰值从 8 万次/日降至稳定 3 万次/日（下降 62.5%）
- API 成本：使用 HolySheheep DeepSeek V3.2 模型，output 价格仅 $0.42/MTok，相同调用量月成本从 $847 降至 $126（节省 85%）
- 循环拦截：系统运行 6 个月内，成功拦截 47 次潜在无限循环事故
- P99 延迟：国内直连 <50ms，用户感知响应时间从平均 3.2 秒降至 0.8 秒
部署时，我把三层防护的阈值设置为：单次请求 Token 上限 8000、最大工具调用深度 5、连续相似响应阈值 85%。这个配置在误拦截率和安全性之间取得了较好平衡。建议你根据实际业务调整这些参数。
六、常见报错排查
报错 1：RATE_LIMITED - QPS 超限
错误信息：{"error": true, "code": "RATE_LIMITED", "message": "QPS 超限，请稍后重试", "retry_after": 0.5}
原因分析：这是令牌桶限流触发的最常见错误。瞬时并发超过桶容量（默认 burst_size=20）时，新请求会被直接拒绝。
解决方案：
# 方案 A：增加突发容量
rate_limiter = MultiLevelRateLimiter(
RateLimitConfig(requests_per_second=10.0, burst_size=50) # 扩容到 50
)
方案 B：实现指数退避重试
async def retry_with_backoff(func, max_retries=3):
for attempt in range(max_retries):
result = await func()
if not result.get("error"):
return result
if result.get("code") == "RATE_LIMITED":
wait_time = result.get("retry_after", 0.5) * (2 ** attempt) # 指数退避
await asyncio.sleep(wait_time)
continue
return result
return {"error": True, "code": "MAX_RETRIES_EXCEEDED"}
使用重试包装
result = await retry_with_backoff(
lambda: client.chat_completion(messages, model="deepseek-v3.2", user_id="user_001")
)报错 2：LOOP_DETECTED - Token 消耗超限
错误信息：{"error": true, "code": "LOOP_DETECTED", "message": "Token 消耗超限: 12500 > 8000"}
原因分析：单次请求的输入+输出 Token 总数超过了阈值，表明 Agent 可能陷入了上下文膨胀循环。
解决方案：
# 方案 A：增加 Token 上限（谨慎使用）
guard = LoopDetectionGuard(
config=LoopDetectionConfig(max_tokens_per_request=15000)
)
方案 B：启用上下文截断
def truncate_messages(messages: list, max_tokens: int = 4000) -> list:
"""智能截断历史消息，保留系统提示和最近对话"""
system_msg = messages[0] if messages and messages[0]["role"] == "system" else None
# 保留最近 N 条消息
recent = messages[-8:] if len(messages) > 8 else messages
if system_msg:
return [system_msg] + recent
return recent
在调用前截断
messages = truncate_messages(conversation_history, max_tokens=4000)
result = await client.chat_completion(messages)报错 3：INFINITE_LOOP - 响应指纹匹配
错误信息：{"error": true, "code": "INFINITE_LOOP", "fingerprint_check": "检测到重复响应 (相似度 91.23%, 累计重复 3 次)"}
原因分析：Agent 连续 3 次生成高度相似的响应，表明陷入了语义层面的循环。
解决方案：
# 方案 A：注入随机扰动
def inject_randomization(system_prompt: str) -> str:
"""在系统提示中注入随机化指令"""
variations = [
"请用不同的表达方式回答",
"尝试换个角度思考这个问题",
"使用更简洁/详细的风格",
]
import random
return system_prompt + f"\n\n[风格要求] {random.choice(variations)}"
修改系统消息
messages[0]["content"] = inject_randomization(messages[0]["content"])
方案 B：重置指纹检测器
client.fingerprint.reset()
result = await client.chat_completion(messages)
方案 C：强制使用工具打断循环
function_call = {
"name": "escalate_to_human",
"arguments": {"reason": "AI 自动检测到循环，升级人工处理"}
}
当检测到循环时，强制输出函数调用
报错 4：CallChainException - 调用深度超限
错误信息：CallChainException: 调用深度超限: 6 > 5 调用链: query_order -> check_inventory -> query_logistics -> check_inventory -> query_logistics...
原因分析：多个工具之间形成了循环依赖链，调用深度超过了预设的 5 层上限。
解决方案：
# 方案 A：识别并修复循环依赖
TOOL_CALL_RULES = {
"check_inventory": {"blocked_after": ["query_logistics"]},
"query_logistics": {"blocked_after": ["check_inventory"]},
}
def prevent_circular_calls(tool_name: str, recent_calls: list) -> bool:
"""阻止会导致循环的工具调用"""
recent_tool_names = [c.split(":")[0] for c in recent_calls[-3:]]
rules = TOOL_CALL_RULES.get(tool_name, {})
for blocked in rules.get("blocked_after", []):
if blocked in recent_tool_names:
return False # 阻止此次调用
return True
在 record_tool_call 中集成检查
if not prevent_circular_calls(tool_name, guard.tool_call_stack):
raise LoopDetectionException(
f"检测到循环依赖: {tool_name} 不能在 {recent_tool_names[-1]} 后调用"
)
方案 B：设置熔断器
class CircuitBreaker:
def __init__(self, threshold: int = 3, timeout: float = 60.0):
self.threshold = threshold
self.timeout = timeout
self.failures = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_failure(self):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.threshold:
self.state = "OPEN"
def is_allowed(self) -> bool:
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
return True
return False
return True
为高风险工具设置熔断器
inventory_circuit = CircuitBreaker(threshold=3,