WeClaw_68_Loop_Detection：检测和中断Agent的死亡循环
第四季系列文章第 7 篇（总第 68 篇） - Loop Detector · 死亡循环 · 三级检测 · 锚定消息注入 · 渐进响应
📚 专栏信息
《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏 · 第四季
专栏定位：面向开发者和技术决策者的实战专栏，用真实案例和完整代码带你理解如何构建生产级 AI 应用
本文深入探讨 Harness Engineering 七大差距中的 G6——Loop Detection。Agent 在 ReAct 循环中可能陷入"死亡循环"：反复调用同一个工具、用相似参数反复尝试同一个操作、在两个操作间来回振荡。本文设计 LoopDetector：一个基于滑动窗口的三级检测器——精确重复、近似重复、振荡模式——配合渐进式三级响应（WARN → REJECT → TERMINATE），在锚定消息中注入循环告警，引导 Agent 换用不同方法。
👨💻 作者与项目
作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念："再复杂的技术，也能用代码讲清楚"
- 💻 项目地址：https://github.com/wyg5208/weclaw.git
- 🌐 官网地址：https://weclaw.link
- 📝 作者 CSDN：https://blog.csdn.net/yweng18
- ⭐ 欢迎 Star⭐、Fork🍴、贡献代码🤝
📝 摘要
本文结构概览：
从 Agent 的"死亡循环"现象出发——模型反复调用 file.write 修改同一个文件，每次参数略有不同，但始终无法成功——分析现有 ExecutionTracker 的参数哈希匹配机制为何不够，然后设计 LoopDetector：一个基于滑动窗口的三级检测器。重点讲解三种循环模式的检测算法（精确重复、近似重复、振荡模式）、渐进式三级响应策略（WARN→REJECT→TERMINATE），以及如何通过 _inject_loop_alert() 在锚定消息中安全注入告警。
核心问题： Agent 在遇到困难时，可能不会换方法，而是反复用同一种方式尝试——每次稍调参数，但核心策略不变。这种"死亡循环"不仅浪费 token，还可能永远无法收敛。如何检测并中断？
关键成果：
- 设计了三级循环检测算法（精确/近似/振荡）
- 渐进式三级响应（WARN→REJECT→终止），第1次仅警告
- 通过锚定消息注入告警，不干扰核心控制流
- 固定内存窗口（8步 × ~200 bytes ≈ 1.6KB/会话）
适合读者：AI Agent 开发者、对 Agent 行为分析感兴趣的工程师
阅读时长：约 14 分钟
关键词：Loop Detection、死亡循环、滑动窗口、振荡检测、锚定消息、渐进响应
一、Agent 的"死亡循环"
1.1 什么是死亡循环？
Agent 在 ReAct 循环中可能陷入三种循环模式：
模式1：精确重复
Step 3: file.write("config.toml", content_v1) → Error: permission denied
Step 4: file.write("config.toml", content_v1) → Error: permission denied ← 完全相同
Step 5: file.write("config.toml", content_v1) → Error: permission denied ← 又来！
Agent 用完全相同的参数反复尝试同一个失败的操作——它"忘记"了上一次失败了。
模式2：近似重复
Step 3: shell.execute("npm install package-a") → Error: network timeout
Step 4: shell.execute("npm install package-a --registry=https://...") → Error: timeout ← 参数微调
Step 5: shell.execute("npm install package-a --registry=https://... --timeout=60000") → Error ← 又微调
Agent 反复尝试同一个操作，每次微调参数，但核心策略不变——它没有意识到需要换一种完全不同的方法。
模式3：振荡模式
Step 3: file.write("a.py", code_v1) → Success
Step 4: shell.execute("python a.py") → Error: syntax error
Step 5: file.write("a.py", code_v2) → Success ← 又开始改
Step 6: shell.execute("python a.py") → Error: different error ← 又开始跑
Step 7: file.write("a.py", code_v3) → Success ← 又改...
Agent 在"修改代码"和"运行测试"之间来回振荡——每次修改引入新错误，修复新错误又引入另一个错误。
1.2 现有机制的不足
WeClaw 的 ExecutionTracker 已经有 is_duplicate_call() 方法：
def is_duplicate_call(self, tool_name, action_name, args_hash) -> bool:
"""检查是否为重复调用（基于参数哈希精确匹配）。"""
key = (tool_name, action_name, args_hash)
if key in self._recent_calls:
return True
self._recent_calls.add(key)
return False
但它只能检测精确重复——近似重复和振荡模式完全无法识别。
1.3 死亡循环的代价
| 维度 | 影响 |
|---|---|
| Token 成本 | 每次循环消耗 500-2000 token，5 次循环 = 2500-10000 token |
| 时间成本 | 每次循环 3-10 秒，5 次循环 = 15-50 秒 |
| 用户体验 | 用户看到 Agent "卡住"了，反复做同一件事 |
| 成功率 | 死亡循环几乎不会收敛——相同策略必然产生相同结果 |
二、LoopDetector 设计
2.1 核心数据结构
@dataclass
class ActionSignature:
"""一次工具调用的签名。"""
tool_name: str
action_name: str
args_hash: str # 参数哈希
args_repr: str # 参数字符串表示（用于相似度计算）
result_status: str # "success" | "error"
step: int # 步骤序号
@dataclass
class LoopReport:
"""循环检测报告。"""
is_looping: bool
pattern_type: str # "exact_repeat" | "approximate_repeat" | "oscillation"
description: str # 人类可读描述
repeat_count: int # 重复次数
affected_steps: list[int] # 涉及的步骤
class LoopDetector:
"""Agent 循环行为检测器。"""
def __init__(self, window_size=8, similarity_threshold=0.85):
self._action_window: deque[ActionSignature] = deque(maxlen=window_size)
self._similarity_threshold = similarity_threshold
self._warn_count = 0 # 警告计数器
2.2 三级检测算法
def detect_loop(self) -> LoopReport:
"""三级检测。"""
if len(self._action_window) < 3:
return LoopReport(is_looping=False, ...)
actions = list(self._action_window)
# 级别1：精确重复检测
report = self._detect_exact_repeat(actions)
if report.is_looping:
return report
# 级别2：近似重复检测
report = self._detect_approximate_repeat(actions)
if report.is_looping:
return report
# 级别3：振荡模式检测
report = self._detect_oscillation(actions)
if report.is_looping:
return report
return LoopReport(is_looping=False, ...)
级别1：精确重复检测
def _detect_exact_repeat(self, actions: list[ActionSignature]) -> LoopReport:
"""检测相同 tool.action + 相同参数哈希的连续重复。"""
# 检查最后 3 个动作是否完全相同
if len(actions) >= 3:
last_3 = actions[-3:]
if (last_3[0].tool_name == last_3[1].tool_name == last_3[2].tool_name and
last_3[0].action_name == last_3[1].action_name == last_3[2].action_name and
last_3[0].args_hash == last_3[1].args_hash == last_3[2].args_hash):
return LoopReport(
is_looping=True,
pattern_type="exact_repeat",
description=f"精确重复: {last_3[2].tool_name}.{last_3[2].action_name} "
f"连续调用 {3} 次，参数完全相同",
repeat_count=3,
affected_steps=[a.step for a in last_3],
)
return LoopReport(is_looping=False, ...)
级别2：近似重复检测
def _detect_approximate_repeat(self, actions: list[ActionSignature]) -> LoopReport:
"""检测相同 tool.action + 相似参数（编辑距离 < 15%）。"""
if len(actions) < 3:
return LoopReport(is_looping=False, ...)
# 取最后 3 个相同 tool.action 的动作
recent = actions[-3:]
if not (recent[0].tool_name == recent[1].tool_name == recent[2].tool_name and
recent[0].action_name == recent[1].action_name == recent[2].action_name):
return LoopReport(is_looping=False, ...)
# 计算参数相似度
sim_01 = _compute_similarity(recent[0].args_repr, recent[1].args_repr)
sim_12 = _compute_similarity(recent[1].args_repr, recent[2].args_repr)
if sim_01 > self._similarity_threshold and sim_12 > self._similarity_threshold:
return LoopReport(
is_looping=True,
pattern_type="approximate_repeat",
description=f"近似重复: {recent[2].tool_name}.{recent[2].action_name} "
f"连续调用 3 次，参数相似度 >{self._similarity_threshold:.0%}",
repeat_count=3,
affected_steps=[a.step for a in recent],
)
return LoopReport(is_looping=False, ...)
def _compute_similarity(s1: str, s2: str) -> float:
"""计算两个字符串的相似度（基于编辑距离）。"""
if not s1 and not s2:
return 1.0
max_len = max(len(s1), len(s2))
if max_len == 0:
return 1.0
distance = _levenshtein(s1, s2)
return 1.0 - distance / max_len
级别3：振荡模式检测
def _detect_oscillation(self, actions: list[ActionSignature]) -> LoopReport:
"""检测 A→B→A→B 交替模式（周期 ≤ 4）。"""
if len(actions) < 4:
return LoopReport(is_looping=False, ...)
# 检查最后 4 个动作是否形成 A→B→A→B 模式
last_4 = actions[-4:]
a_sig = (last_4[0].tool_name, last_4[0].action_name)
b_sig = (last_4[1].tool_name, last_4[1].action_name)
if (a_sig == (last_4[2].tool_name, last_4[2].action_name) and
b_sig == (last_4[3].tool_name, last_4[3].action_name) and
a_sig != b_sig):
return LoopReport(
is_looping=True,
pattern_type="oscillation",
description=f"振荡模式: {a_sig[0]}.{a_sig[1]} ↔ {b_sig[0]}.{b_sig[1]} "
f"交替循环",
repeat_count=2,
affected_steps=[a.step for a in last_4],
)
return LoopReport(is_looping=False, ...)
三、渐进式三级响应
3.1 响应策略
LoopDetector 不一次性终止循环——它采用渐进式三级响应：
第1次检测到循环 → WARN：在锚定消息中追加告警文本
第2次检测到循环 → REJECT：跳过当前工具调用，向 _batch 追加拒绝消息
第3次检测到循环 → TERMINATE：构建部分结果，终止循环
3.2 为什么不直接终止？
直接终止的问题：
- 可能误报——有时候"重复"是合理的（比如分步写入大文件）
- 用户会困惑——Agent 突然停止了，不知道为什么
渐进式响应给了 Agent 修正机会：
- WARN 提示它"你可能在做重复的事"
- 如果 Agent 仍然重复，REJECT 强制跳过
- 如果还在重复，TERMINATE 才终止
3.3 实现逻辑
def get_response_action(self, report: LoopReport) -> str:
"""根据循环检测报告和警告次数决定响应动作。"""
if not report.is_looping:
self._warn_count = 0 # 重置
return "PASS"
self._warn_count += 1
if self._warn_count == 1:
return "WARN" # 在锚定消息中追加告警
elif self._warn_count == 2:
return "REJECT" # 跳过当前工具调用
else:
return "TERMINATE" # 终止循环
四、Hook 注入 —— 锚定消息中的循环告警
4.1 注入位置
LoopDetector 的告警注入在锚定消息中——这是 ReAct 循环每一步开始时构建的"上下文提示"：
def _inject_loop_alert(self, anchor_content: str, session_id: str) -> str:
"""在锚定消息中注入循环告警（G6），返回修改后的内容。"""
if not self._loop_detector:
return anchor_content
report = self._loop_detector.detect_loop()
if not report.is_looping:
return anchor_content
action = self._loop_detector.get_response_action(report)
if action == "WARN":
alert = (
f"\n\n[循环告警] {report.description}\n"
f"请停止当前操作模式，换用不同方法或直接报告当前结果。"
)
# 旁路通知（不阻塞主流程）
asyncio.ensure_future(
self.event_bus.emit(
EventType.LOOP_DETECTED,
LoopDetectedEvent(
session_id=session_id,
pattern=report.pattern_type,
count=report.repeat_count,
)
)
)
return anchor_content + alert
return anchor_content
4.2 双路径注入
# _chat_impl 路径（L1539）
anchor_content = self._build_smart_anchor_message(...)
anchor_content = self._inject_loop_alert(anchor_content, session_id) # ← 注入
# chat_stream 路径（L2195）
anchor_content = self._build_smart_anchor_message(...)
anchor_content = self._inject_loop_alert(anchor_content, session_id) # ← 注入
4.3 REJECT 和 TERMINATE 的安全处理
# REJECT：在工具执行循环中跳过当前工具
if action == "REJECT":
_batch.append(("tool",
f"[循环检测] 工具调用被拒绝: 重复模式 {report.pattern_type}",
{"tool_call_id": tc_entry["id"]}))
continue # 跳过这个工具调用
# TERMINATE：构建部分结果并退出
if action == "TERMINATE":
partial_result = (
f"任务因检测到持续循环而终止。\n"
f"循环模式: {report.description}\n"
f"已完成步骤: {step_idx}"
)
_batch.append(("assistant", partial_result, {}))
return # 退出 ReAct 循环
五、EventBus 集成
5.1 新增事件类型
# events.py
class EventType:
# ... 现有 31 个事件 ...
LOOP_DETECTED = "loop_detected" # 新增
@dataclass
class LoopDetectedEvent:
session_id: str
pattern: str # "exact_repeat" | "approximate_repeat" | "oscillation"
count: int
5.2 事件订阅
其他模块可以订阅 LOOP_DETECTED 事件来做旁路处理：
# 审计日志自动记录
audit_logger.subscribe(EventType.LOOP_DETECTED, lambda event:
audit_logger.log(
action="loop_detected",
details={"pattern": event.pattern, "count": event.count}
)
)
# TaskTrace 记录
trace_collector.subscribe(EventType.LOOP_DETECTED, lambda event:
trace_collector.add_event("loop_detected", event.__dict__)
)
EventBus 的作用是旁路通知——不阻塞主流程。告警文本已经通过 _inject_loop_alert 注入到了锚定消息中，EventBus 事件只是给审计和追踪系统用的。
六、内存与性能
6.1 内存占用
self._action_window: deque[ActionSignature] = deque(maxlen=8)
固定窗口大小 8，每个 ActionSignature 约 200 bytes：
tool_name: ~20 bytesaction_name: ~20 bytesargs_hash: ~32 bytes (MD5)args_repr: ~100 bytes (截断)result_status: ~10 bytesstep: ~8 bytes (int)
总内存：8 × 200 = 1.6 KB / 会话。这是可以忽略的内存开销。
6.2 检测延迟
每次 detect_loop() 需要遍历窗口中的 8 个动作：
- 精确重复：3 次比较 = O(1)
- 近似重复：2 次编辑距离计算 = O(n²)，但 n 是 args_repr 长度（通常 <200），实际 <1ms
- 振荡检测：4 次比较 = O(1)
总延迟：<2ms，对 ReAct 循环的每步 3-10 秒来说可以忽略。
6.3 编辑距离优化
对于长参数字符串，可以使用更快的相似度算法：
def _compute_similarity_fast(s1: str, s2: str) -> float:
"""快速相似度计算（基于字符集交集）。"""
if not s1 and not s2:
return 1.0
set1, set2 = set(s1), set(s2)
intersection = set1 & set2
union = set1 | set2
return len(intersection) / len(union) if union else 1.0
Jaccard 相似度的计算复杂度是 O(n)，比编辑距离的 O(n²) 快很多。虽然精度略低，但对于"检测近似重复"已经足够。
七、配置与灰度
[agent.loop_detection]
enabled = false
repeat_threshold = 3 # 连续重复次数阈值
similarity_threshold = 0.85 # 近似重复相似度阈值
window_size = 8 # 滑动窗口大小
intervention = "anchor_message" # 干预方式
7.1 灰度策略
- 第一周：仅启用 WARN（第1级），不 REJECT 不 TERMINATE
- 第二周：追加 REJECT（第2级），观察误拦截率
- 第三周：追加 TERMINATE（第3级），完整三级响应
7.2 误报监控
# 每次循环检测触发时记录
logger.info(
"LoopDetector 触发: pattern=%s, action=%s, warn_count=%d, "
"tool=%s, step=%d",
report.pattern_type, action, self._warn_count,
actions[-1].tool_name, actions[-1].step
)
通过审计日志统计误报率（WARN 后 Agent 是否成功改变策略），如果误报率 >30%，调高 similarity_threshold。
八、核心教训
8.1 检测和响应要分离
LoopDetector 的检测逻辑（三级算法）和响应逻辑（渐进三级）是分离的。检测只负责"发现循环"，响应决定"怎么处理"。这种分离使得我们可以独立调整检测灵敏度和响应力度。
8.2 锚定消息是最好的告警通道
WeClaw 的 _build_smart_anchor_message() 是 ReAct 循环每步开始时构建的上下文提示——它在模型调用之前被注入到上下文中。在锚定消息中追加循环告警，模型会在下一步推理时"看到"这个告警，从而有机会调整策略。
这比直接 continue 跳过工具调用更温和——模型仍然可以执行操作，但它知道了"你在重复"。
8.3 振荡检测是最有价值的
精确重复和近似重复相对容易发现——连续 3 次做同样的事。但振荡模式（A→B→A→B）更隐蔽——Agent 看起来在做不同的事（A 和 B 不同），但整体在"打转"。振荡检测是 LoopDetector 区别于现有 ExecutionTracker 的核心价值。
📖 相关文章
本文是 WeClaw 专栏第四季的第 7 篇（总第 68 篇）。如果这篇文章对你有帮助，欢迎给项目点个 Star ⭐