浏览器智能体可观测性（一）：操作流水

为什么可观测性是刚需

你的智能体在生产环境中跑了 50 步操作——导航、填表、搜索、提取。然后它报错了。或者更糟：它没报错，但结果错了。

你怎么知道它在哪里出错的？

传统的日志系统在这个场景下基本失效。HTTP 请求日志告诉你请求发了、响应收到了，但不知道智能体当时在看页面的哪个部分。应用日志告诉你智能体调用了哪个函数，但不知道它为什么调用。

AI 浏览器智能体的可观测性需要记录的信息比传统系统多得多：

每步的操作内容——智能体决定做什么
执行时的页面状态——智能体看到了什么
决策的原因——智能体为什么做这个决定
操作的结果——页面发生了什么变化

没有这些信息，调试 AI 浏览器自动化就是拿着手电筒在黑暗里找一根针。

操作流水日志的最小记录单元

每个操作步骤至少记录以下信息：

class StepLog:
    """单个操作步骤的完整记录"""
    def __init__(self, step_number, action, reasoning, page_state):
        self.step_number = step_number
        self.action = action              # 智能体执行的操作
        self.reasoning = reasoning        # 智能体的决策原因
        self.timestamp = datetime.now()
        self.page_state = page_state      # 操作前的页面状态
        self.result = None                # 操作结果
        self.error = None                 # 错误信息
        self.duration_ms = 0              # 操作耗时
 
class PageState:
    """操作时的页面状态快照"""
    def __init__(self, url, title, screenshot_path, dom_snapshot):
        self.url = url
        self.title = title
        self.screenshot_path = screenshot_path
        self.dom_snapshot = dom_snapshot
        self.cookies = None

三种记录级别的成本

页面状态快照有不同的粒度，粒度和存储成本呈正比：

记录级别	每步数据量	1000 步的任务	适合场景
仅操作日志	~100 bytes	~100 KB	低成本监控，大批量任务
+ DOM 快照	~10 KB	~10 MB	调试常见错误
+ 全页截图	~500 KB	~500 MB	复杂场景的完整回放

建议：默认使用 DOM + 操作日志，只有被标记为失败的任务才保留截图。

实施：在 agent-browser 中记录操作流

import json, os, time, base64
from datetime import datetime
 
class AgentTracer:
    def __init__(self, task_id, output_dir="./traces"):
        self.task_id = task_id
        self.trace_dir = os.path.join(output_dir, task_id)
        os.makedirs(self.trace_dir, exist_ok=True)
        self.steps = []
        self.start_time = time.time()
 
    async def record_step(self, agent, action_result):
        step = {
            "step": len(self.steps) + 1,
            "timestamp": datetime.now().isoformat(),
            "action": action_result["action"],
            "reasoning": action_result.get("reasoning", ""),
            "url": agent.page.url,
            "title": await agent.page.title(),
            "dom_snapshot": await self.get_dom_snapshot(agent),
        }
        self.steps.append(step)
 
        # 每 10 步写一次到磁盘，避免内存占用过高
        if len(self.steps) % 10 == 0:
            self.flush()
 
    async def get_dom_snapshot(self, agent):
        """获取交互元素的 A11y 快照"""
        return await agent.page.evaluate("""
            () => {
                const elements = document.querySelectorAll(
                    'button, a, input, select, textarea, [role="button"]'
                );
                return Array.from(elements).slice(0, 50).map(el => ({
                    tag: el.tagName,
                    text: el.textContent?.trim().slice(0, 100),
                    visible: el.offsetParent !== null
                }));
            }
        """)
 
    def flush(self):
        path = os.path.join(self.trace_dir, "steps.jsonl")
        with open(path, "a") as f:
            for step in self.steps[-10:]:
                f.write(json.dumps(step) + "\n")

回放机制

记录操作流水是为了回放。回放 Debugger 应该允许你：

按步骤前进/后退——看智能体每一步看到的页面状态
查看 LLM 的完整推理过程——智能体为什么点击这个按钮而不是那个
对比预期结果和实际结果——智能体认为操作成功了吗？实际上呢？

class TraceReplay:
    def __init__(self, trace_path):
        self.steps = self.load_trace(trace_path)
 
    def load_trace(self, path):
        steps = []
        with open(path) as f:
            for line in f:
                if line.strip():
                    steps.append(json.loads(line))
        return steps
 
    def get_step(self, n):
        """获取第 N 步的完整信息"""
        step = self.steps[n]
        return {
            "step": step["step"],
            "before": {
                "url": step["url"],
                "title": step["title"],
                "dom": step["dom_snapshot"],
            },
            "action": step["action"],
            "reasoning": step["reasoning"],
            "after": {
                "result": step.get("result"),
                "error": step.get("error"),
            }
        }
 
    def find_deviations(self):
        """找出操作未产生预期页面变化的步骤"""
        deviations = []
        for i, step in enumerate(self.steps):
            if step.get("action_type") == "click":
                # 如果点击后 URL 没变且 DOM 没变，点击可能没生效
                if i + 1 < len(self.steps):
                    next_step = self.steps[i + 1]
                    if (step["url"] == next_step["url"] and
                        step["dom_snapshot"] == next_step["dom_snapshot"]):
                        deviations.append(i)
        return deviations

find_deviations() 方法找出那些"点了但什么都没变化的"步骤——这是 Nanobrowser Validator 幻觉（认为操作成功但实际失败）最常见的情况。

发现异常行为模式

通过对操作流水数据的分析，可以自动发现 AI 智能体的异常行为模式：

class AnomalyDetector:
    def __init__(self, trace):
        self.trace = trace
 
    def detect_loops(self):
        """检测重复点击同一位置的循环"""
        actions = [(s["url"], s["action"].get("selector"))
                   for s in self.trace.steps]
        for i in range(len(actions) - 3):
            if actions[i] == actions[i+1] == actions[i+2]:
                return {"loop_detected": True,
                        "start_step": i, "action": actions[i]}
        return {"loop_detected": False}
 
    def detect_cost_anomaly(self):
        """检测 Token 消耗异常"""
        token_counts = [s.get("tokens_consumed", 0) for s in self.trace.steps]
        avg_tokens = sum(token_counts) / len(token_counts) if token_counts else 0
        # 如果某一步的消耗超过平均值 3 倍
        for i, count in enumerate(token_counts):
            if avg_tokens > 0 and count > avg_tokens * 3:
                return {"spike_at_step": i,
                        "tokens": count, "avg": avg_tokens}
        return None

总结

没有操作流水数据的 AI 浏览器自动化是盲人开车。最基本的可观测性要求是：

每步记录——操作、推理、页面状态
可回放——失败时可以回溯每一步的上下文
自动检测——通过分析流水数据发现循环、成本异常和无效操作

下一篇文章讨论如何通过回放 Debugger 像看录像一样查看智能体的每一步操作。

AI 浏览器智能体可观测性（一）：操作流水——智能体每一步都干了什么