AI 浏览器智能体可观测性（二）：监控与告警——HTTP 200 不代表任务成功

# 智能体的一个"成功"的 HTTP 请求
HTTP 200, body: {"status": "ok"}  
# 但智能体在上一步提取数据时，因为页面改版没找到价格字段
# 它重试了 10 次，每次 HTTP 都返回 200，每次都找不到

HTTP 200 + 空数据
  → 页面加载成功，但目标元素不存在
  → 监控认为正常，业务认为失败
 
P95 延迟正常 + 步数超标
  → 单次请求不慢，但智能体在同一步循环了 20 次
  → 监控认为正常，实际任务耗时翻了 4 倍
 
无 5xx + Validator 幻觉
  → 服务器没有报错，但 Validator 误判了任务状态
  → 监控认为正常，实际上操作未生效

class TaskOutcome:
    """任务的真实结果，不是 HTTP 状态码"""
    SUCCESS = "success"           # 数据完整，操作生效
    DEGRADED = "degraded"         # 成功但有降级（命中缓存、跳过可选字段）
    EXTRACTION_EMPTY = "empty"    # 页面加载了但目标数据是空的
    LOOP_DETECTED = "loop"        # 同一步操作循环了 N 次
    TIMEOUT = "timeout"           # 超过时间限制
    VALIDATOR_ERROR = "hallucination"  # Validator 报告成功但实际未生效

class LoopDetector:
    """检测操作序列中是否存在重复模式"""
    def __init__(self, window=5):
        self.window = window
 
    def check(self, action_history):
        if len(action_history) < self.window * 2:
            return False, 0.0
        recent = action_history[-self.window:]
        previous = action_history[-self.window*2:-self.window]
        similarity = self.calc_similarity(recent, previous)
        is_looping = similarity >= 0.8
        return is_looping, similarity
 
    def calc_similarity(self, a, b):
        """计算两组操作的动作相似度"""
        matches = sum(1 for x, y in zip(a, b)
                      if x.get("action") == y.get("action")
                      and x.get("selector") == y.get("selector"))
        return matches / max(len(a), 1)

class StepAnomalyDetector:
    """通过步数偏离检测智能体是否遇到困难"""
    def __init__(self):
        self.baseline_steps = {}  # task_type -> median_steps
 
    def record_step(self, task_type, step_count, success):
        if task_type not in self.baseline_steps:
            self.baseline_steps[task_type] = []
        self.baseline_steps[task_type].append(step_count)
 
    def is_anomalous(self, task_type, current_steps):
        baseline = self.baseline_steps.get(task_type, [])
        if len(baseline) < 10:
            return False  # 数据不够，不做判断
        median = sorted(baseline)[len(baseline) // 2]
        # 超过基线 3 倍视为异常
        return current_steps > median * 3, median

class AlertManager:
    def __init__(self):
        self.alerts = []
 
    def evaluate(self, task_outcome):
        """根据任务结果决定告警级别和渠道"""
        if task_outcome == "loop":
            # 循环卡住：发到工作群，但不 pager
            self.notify_slack(f"任务进入循环: {task_outcome.task_id}")
            return "warning"
 
        if task_outcome == "extraction_empty":
            # 提取为空：自动重试，连续 fail 再告警
            self.record_empty(task_outcome.task_id)
            if self.consecutive_empties >= 3:
                self.notify_slack(f"连续 3 次提取为空: {task_outcome.url}")
                return "warning"
            return "info"
 
        if task_outcome == "validator_error":
            # Validator 误判：需要人工确认数据的正确性
            self.notify_pager(f"可能的数据错误: {task_outcome.task_id}")
            return "critical"
 
        return "ok"

class CostTracker:
    def __init__(self, budget_per_task=0.10, daily_budget=50.0):
        self.budget_per_task = budget_per_task
        self.daily_budget = daily_budget
        self.daily_total = 0.0
        self.spikes = []
 
    def record_task(self, task_id, tokens, model_price):
        cost = (tokens / 1000) * model_price
        self.daily_total += cost
 
        if cost > self.budget_per_task:
            self.spikes.append((task_id, cost))
            self.notify_spike(task_id, cost)
 
    def check_daily(self):
        if self.daily_total > self.daily_budget:
            self.halt_new_tasks(f"日预算超限: {self.daily_total}")