老系统自动化的容错机制:当UI界面发生微调,Agent如何自愈?
一、UI微调为什么会让Agent“翻车”
老系统的UI变化,通常不是“改版”,而是“微调”:
- 按钮从坐标(320, 180)挪到了(340, 180)
- 标签文案从“查询”改成了“查询(Q)”
- 输入框从标准控件换成了自绘控件
- 弹窗从一层变成了两层嵌套
- 表头列的顺序调整了
对依赖“固定坐标点击”的RPA脚本,任何一个微调都会导致点击落空或点错位置。对依赖“固定文本匹配”的脚本,文案变化会让元素定位失败。
问题的本质是:Agent对UI的“认知”是脆弱的。 它记住的是“坐标”或“精确文本”,而不是“这个控件的语义角色”。UI一微调,认知就失效。
下图对比了“脆弱认知”与“语义认知”两种定位方式的差异:
二、容错架构:四层防御 + 状态机恢复
┌─────────────────────────────────────────────────────────────────┐
│ Agent执行引擎 │
│ 任务规划 → 元素定位 → 操作执行 → 结果验证 → 状态更新 │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第一层:变化检测层(Change Detection) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 界面指纹比对 │ │ 元素基线偏移 │ │ 布局结构变化 │ │
│ │ 截图哈希 │ │ 位置/文本 │ │ DOM/控件树 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第二层:元素定位回退层(Locator Fallback) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 精确匹配 │ │ 模糊匹配 │ │ 语义匹配 │ │
│ │ 坐标+文本 │ │ 正则/相似度 │ │ 控件角色+上下文│ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第三层:状态机恢复层(State Machine Recovery) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 检查点持久化 │ │ 断点回退 │ │ 幂等重执行 │ │
│ │ 步骤级快照 │ │ 回到安全状态 │ │ 防重复副作用 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ 第四层:熔断止损层(Circuit Breaker) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 连续失败计数 │ │ 自动降级 │ │ 人工接管 │ │
│ │ 阈值熔断 │ │ 切换备用策略 │ │ 升级通知 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
核心设计原则:先检测变化,再回退定位;定位失败走状态恢复,恢复不了走熔断止损。每一层降低一部分风险,叠加起来才能让Agent在UI微调后“自愈”。
四层防御的协作关系可以用下面的流程图表示:
三、第一层:变化检测——先知道“哪里变了”
Agent在执行操作前,先做界面指纹比对。如果指纹没变,走快速路径;如果指纹变了,进入回退定位流程。
import hashlib
from dataclasses import dataclass
from typing import Optional
@dataclass
class UIFingerprint:
"""界面指纹:用于检测UI是否发生变化"""
screenshot_hash: str # 截图感知哈希
element_baseline: dict # 关键元素的位置和文本基线
layout_signature: str # 布局结构签名(控件树序列化哈希)
class ChangeDetector:
"""变化检测器:比对当前界面与基线"""
def __init__(self, baseline_store, similarity_threshold=0.95):
self.baseline = baseline_store
self.threshold = similarity_threshold
def detect(self, screen_id: str, current: UIFingerprint) -> dict:
baseline = self.baseline.get(screen_id)
if not baseline:
# 首次执行,建立基线
self.baseline.save(screen_id, current)
return {"changed": False, "reason": "baseline_created"}
# 1. 截图感知哈希比对(对亮度、微小位移不敏感)
hash_sim = self._hamming_similarity(
baseline.screenshot_hash, current.screenshot_hash
)
# 2. 关键元素偏移检测
element_shifts = []
for elem_id, base in baseline.element_baseline.items():
curr = current.element_baseline.get(elem_id)
if not curr:
element_shifts.append({
"element": elem_id, "type": "missing"
})
elif curr["text"] != base["text"]:
element_shifts.append({
"element": elem_id, "type": "text_changed",
"from": base["text"], "to": curr["text"]
})
elif self._position_shift(base["bbox"], curr["bbox"]) > 20:
element_shifts.append({
"element": elem_id, "type": "position_shift",
"delta": self._position_shift(base["bbox"], curr["bbox"])
})
# 3. 布局结构变化
layout_changed = baseline.layout_signature != current.layout_signature
changed = hash_sim < self.threshold or element_shifts or layout_changed
return {
"changed": changed,
"hash_similarity": round(hash_sim, 3),
"element_shifts": element_shifts,
"layout_changed": layout_changed,
"strategy": "fallback_locator" if changed else "fast_path"
}
def _hamming_similarity(self, h1: str, h2: str) -> float:
"""感知哈希的汉明距离相似度"""
if len(h1) != len(h2):
return 0.0
diff = sum(1 for a, b in zip(h1, h2) if a != b)
return 1.0 - diff / len(h1)
def _position_shift(self, bbox1: dict, bbox2: dict) -> float:
"""计算元素位置偏移量"""
cx1 = bbox1["x"] + bbox1["w"] / 2
cy1 = bbox1["y"] + bbox1["h"] / 2
cx2 = bbox2["x"] + bbox2["w"] / 2
cy2 = bbox2["y"] + bbox2["h"] / 2
return ((cx1 - cx2) ** 2 + (cy1 - cy2) ** 2) ** 0.5
关键设计:感知哈希对“亮度微变、1-2像素位移”不敏感,但对“布局重构、元素消失”敏感。元素级比对则精确到“哪个元素、什么类型的变化”。两者结合,既能避免误报,又能发现真实变化。
变化检测的判定逻辑如下:
四、第二层:元素定位回退——从“精确匹配”到“语义匹配”
检测到变化后,Agent不能直接放弃,而要逐级回退定位策略。
from enum import Enum
from typing import Optional
class LocatorStrategy(Enum):
EXACT = "exact" # 精确匹配:坐标+文本
FUZZY_TEXT = "fuzzy_text" # 模糊文本:正则/相似度
SEMANTIC = "semantic" # 语义匹配:控件角色+上下文
RELATIVE = "relative" # 相对定位:基于邻近元素
class LocatorFallback:
"""元素定位回退链:逐级降级,直到找到目标"""
def __init__(self, vision_model, ocr_engine, llm):
self.vision = vision_model
self.ocr = ocr_engine
self.llm = llm
def locate(self, screen: bytes, target: dict) -> dict:
"""按回退链定位元素"""
# 策略1:精确匹配(坐标+文本)
result = self._exact_match(screen, target)
if result:
return {"found": True, "strategy": "exact", "element": result}
# 策略2:模糊文本匹配
result = self._fuzzy_text_match(screen, target)
if result:
return {"found": True, "strategy": "fuzzy_text", "element": result}
# 策略3:语义匹配(控件角色+上下文)
result = self._semantic_match(screen, target)
if result:
return {"found": True, "strategy": "semantic", "element": result}
# 策略4:相对定位(基于邻近元素)
result = self._relative_match(screen, target)
if result:
return {"found": True, "strategy": "relative", "element": result}
return {"found": False, "strategy": "none"}
def _exact_match(self, screen, target) -> Optional[dict]:
"""精确匹配:坐标和文本都匹配"""
elements = self.vision.detect(screen)
for e in elements:
if (abs(e["bbox"]["x"] - target["bbox"]["x"]) < 5
and e["label"] == target["label"]):
return e
return None
def _fuzzy_text_match(self, screen, target) -> Optional[dict]:
"""模糊文本匹配:文本相似度 > 0.8"""
elements = self.vision.detect(screen)
text_blocks = self.ocr.detect(screen)
for e in elements:
label = self._match_label(e["bbox"], text_blocks)
if label and self._similarity(label, target["label"]) > 0.8:
return e
return None
def _semantic_match(self, screen, target) -> Optional[dict]:
"""语义匹配:用LLM判断控件角色是否匹配"""
elements = self.vision.detect(screen)
text_blocks = self.ocr.detect(screen)
candidates = []
for e in elements:
label = self._match_label(e["bbox"], text_blocks)
candidates.append({
"type": e["type"],
"label": label,
"bbox": e["bbox"]
})
prompt = f"""
目标元素:类型={target['type']},功能={target['label']}
界面候选元素:{candidates}
请找出与目标元素语义最匹配的候选(考虑控件类型和功能描述)。
输出JSON:{{"index": 0, "confidence": 0.0-1.0}}
"""
import json
result = json.loads(self.llm.generate(prompt))
if result["confidence"] > 0.7:
return candidates[result["index"]]
return None
def _relative_match(self, screen, target) -> Optional[dict]:
"""相对定位:基于邻近已知元素"""
# 如果目标元素附近有已知的锚点元素,用相对位置定位
anchor = target.get("anchor_element")
if not anchor:
return None
anchor_elem = self._exact_match(screen, anchor)
if not anchor_elem:
return None
# 基于锚点偏移计算目标位置
dx = target["bbox"]["x"] - anchor["bbox"]["x"]
dy = target["bbox"]["y"] - anchor["bbox"]["y"]
return {
"bbox": {
"x": anchor_elem["bbox"]["x"] + dx,
"y": anchor_elem["bbox"]["y"] + dy,
"w": target["bbox"]["w"],
"h": target["bbox"]["h"]
},
"label": target["label"],
"confidence": 0.6
}
关键设计:回退链是“逐级降级”的——精确匹配最快最准,语义匹配最慢但最鲁棒。每一级失败才进入下一级,避免“一上来就用最慢的策略”。
定位回退链的逐级降级过程:
五、第三层:状态机恢复——从“失败步骤”回到“安全状态”
定位失败或操作失败后,Agent需要从当前状态回退到最近的安全检查点,而不是从头开始。
from enum import Enum
from dataclasses import dataclass, field
from typing import List, Optional
class StepState(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
ROLLED_BACK = "rolled_back"
@dataclass
class Checkpoint:
"""检查点:步骤级快照"""
step_id: str
state: StepState
ui_snapshot: bytes # 执行前的界面截图
data_snapshot: dict # 执行前的数据状态
compensations: List[dict] # 补偿操作(用于回滚)
class StateMachineRecovery:
"""状态机恢复:检查点 + 回滚 + 幂等重执行"""
def __init__(self, checkpoint_store, executor):
self.checkpoints = checkpoint_store
self.executor = executor
def save_checkpoint(self, task_id: str, step_id: str,
ui_snapshot: bytes, data_snapshot: dict):
"""步骤执行前保存检查点"""
self.checkpoints.save(task_id, Checkpoint(
step_id=step_id,
state=StepState.RUNNING,
ui_snapshot=ui_snapshot,
data_snapshot=data_snapshot,
compensations=[]
))
def recover(self, task_id: str, failed_step: str) -> dict:
"""从失败步骤恢复"""
checkpoint = self.checkpoints.get_latest_success(task_id)
if not checkpoint:
return {"action": "restart", "reason": "no_checkpoint"}
# 1. 回滚失败步骤可能产生的副作用
for comp in checkpoint.compensations:
self.executor.execute_compensation(comp)
# 2. 恢复界面状态
self._restore_ui(checkpoint.ui_snapshot)
# 3. 从检查点继续
return {
"action": "resume",
"resume_from": checkpoint.step_id,
"data_state": checkpoint.data_snapshot
}
def _restore_ui(self, snapshot: bytes):
"""恢复界面:关闭多余弹窗,回到已知状态"""
# 发送ESC关闭弹窗,回到主界面
self.executor.send_key("ESC")
# 验证界面与快照一致
current = self.executor.capture()
if not self._ui_matches(current, snapshot):
# 如果界面不一致,强制刷新
self.executor.send_key("F5")
关键设计:检查点保存的是“执行前的状态”,不是“执行后的状态”。失败后回退到“执行前”,再重试。补偿操作用于撤销已经产生的副作用(如已提交的表单),保证重试的幂等性。
步骤状态与恢复路径的状态机如下:
六、第四层:熔断止损——当自愈也失败时
自愈不是万能的。连续失败达到阈值,Agent必须停下来,而不是“无限重试”。
class UIAutomationCircuitBreaker:
"""UI自动化熔断器:连续失败自动止损"""
def __init__(self, failure_threshold=3, recovery_timeout=300):
self.failure_count = 0
self.threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = "closed" # closed | open | half_open
self.last_failure_time = 0
def on_result(self, success: bool):
if success:
self.failure_count = 0
if self.state == "half_open":
self.state = "closed"
else:
self.failure_count += 1
self.last_failure_time = now_ms()
if self.failure_count >= self.threshold:
self.state = "open"
def allow_attempt(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
# 恢复窗口后进入半开状态,放行一次探测
if now_ms() - self.last_failure_time > self.recovery_timeout * 1000:
self.state = "half_open"
return True
return False
# half_open:只放行一次探测
return True
def get_fallback_action(self) -> dict:
"""熔断后的降级动作"""
if self.state == "open":
return {
"action": "escalate_to_human",
"reason": "ui_automation_unstable",
"message": "连续操作失败,已暂停自动执行,转人工处理"
}
return {"action": "continue"}
关键设计:熔断后不是“永久停用”,而是“打开—半开—闭合”的循环。半开状态放行一次探测,成功则恢复,失败则继续熔断。这和分布式系统的熔断器逻辑一脉相承。
七、完整执行流程伪代码
async def execute_with_resilience(task, agent_context):
breaker = UIAutomationCircuitBreaker()
detector = ChangeDetector(baseline_store)
locator = LocatorFallback(vision, ocr, llm)
recovery = StateMachineRecovery(checkpoint_store, executor)
for step in task.steps:
# 1. 熔断检查
if not breaker.allow_attempt():
return breaker.get_fallback_action()
# 2. 界面变化检测
current_fp = capture_fingerprint()
change = detector.detect(step.screen_id, current_fp)
# 3. 元素定位(按需回退)
if change["strategy"] == "fallback_locator":
target = locator.locate(current_screen, step.target)
if not target["found"]:
breaker.on_result(False)
recovered = recovery.recover(task.id, step.id)
if recovered["action"] == "restart":
breaker.on_result(False)
return {"status": "failed", "reason": "locator_failed"}
continue
# 4. 保存检查点后执行
recovery.save_checkpoint(task.id, step.id,
capture_screen(), get_data_state())
try:
execute_action(target)
breaker.on_result(True)
except Exception:
breaker.on_result(False)
recovered = recovery.recover(task.id, step.id)
if recovered["action"] != "resume":
return {"status": "failed", "reason": "execution_failed"}
return {"status": "success"}
八、工程实践要点
基线要定期更新。 UI微调是渐进的,基线不能“一次建立、永久使用”。建议在成功执行后,对比当前界面与基线,差异在阈值内则更新基线,差异过大则告警。
回退链要有代价意识。 精确匹配毫秒级,语义匹配可能几百毫秒。回退链的顺序应该是“从快到慢”,而不是“从准到不准”。
检查点存“执行前”状态。 存执行后的状态,回滚时无法撤销副作用。存执行前的状态,才能“回到起点重来”。
熔断阈值按操作风险分级。 只读操作可以容忍更多重试,写入操作应该更低阈值——写错数据的代价远高于读失败。
人工接管要带上下文。 熔断后转人工,必须把“执行到哪一步、失败原因、已产生的副作用”一并推送,而不是只发一条“Agent失败了”。
在行业实践中,类似沈管家这类全自动AI数字员工的任务执行引擎,已将状态持久化、断点恢复和异常分级处理作为执行层的基础能力——任务执行有检查点,失败可从断点继续,连续异常触发熔断并转人工。这种“检查点+回退+熔断”的分工,是UI自动化在真实老系统环境中能够稳定运行的工程前提。
九、结语
老系统的UI微调不可避免,Agent的失败也不可避免。真正的容错能力,不是“永远不失败”,而是“失败后能检测、能回退、能恢复、能止损”。
变化检测让Agent“知道哪里变了”,定位回退让Agent“换个方式找元素”,状态机恢复让Agent“从安全状态重来”,熔断机制让Agent“知道什么时候该停”。四层叠加,Agent才能从“演示时能跑”走向“生产环境能扛”。
你们在UI自动化中遇到过哪些“界面一改就翻车”的情况?是坐标失效、文本变化,还是弹窗干扰?欢迎在评论区交流你们的容错方案和踩坑经验。
更多推荐



所有评论(0)