Playwright + AI 智能体混合架构(二):桥接层设计——让脚本和 AI 自由切换

同一套任务里,脚本做导航和提取,AI 处理异常和适配。桥接层负责协调两者、统一接口、保证切换透明。

亿牛云技术团队2026年4月16日3 分钟阅读

桥接层要解决的核心问题

混合架构的第一个问题(D1 篇)是"该用脚本还是 AI",第二个问题是"怎么让它们协作"。

你有一个任务:"登录电商网站 → 搜索商品 → 提取价格 → 提交到比价系统。"其中搜索和提取的页面结构稳定,用脚本做最好。但登录可能出验证码,提取时可能页面改版。

需要的是:同一套任务流中,某些步骤用脚本,某些步骤用 AI,出错时在两者之间透明切换。

统一接口抽象

桥接层的核心是一个统一的操作接口——不管背后是脚本实现还是 AI 实现,调用方不需要知道。

from abc import ABC, abstractmethod
 
class BrowserOperation(ABC):
    """统一的操作接口"""
    @abstractmethod
    async def execute(self, page, context):
        pass
 
class ScriptOperation(BrowserOperation):
    """脚本实现"""
    async def execute(self, page, context):
        await page.goto(context["url"])
        await page.fill("#search", context["keyword"])
        await page.click("#search-btn")
        await page.wait_for_selector(".results")
        return await page.text_content(".results")
 
class AIOperation(BrowserOperation):
    """AI 实现"""
    async def execute(self, page, context):
        task = f"在 {context['url']} 搜索 '{context['keyword']}' 并返回结果"
        return await self.agent.run(task)

调用方不需要关心具体的实现。下一步是决定什么时候用哪种实现。

错误检测与切换逻辑

桥接层的第二个核心功能是:检测脚本操作是否失败,并在失败时自动切换到 AI。

class HybridExecutor:
    def __init__(self):
        self.fallback_threshold = 0
 
    async def execute_with_fallback(self, operation, page, context):
        # 先用脚本
        try:
            result = await ScriptOperation().execute(page, context)
            return result
        except (TimeoutError, SelectorNotFoundError, NetworkError) as e:
            self.fallback_threshold += 1
            # 切换到 AI
            logger.info(f"Script failed ({str(e)}), falling back to AI")
            result = await AIOperation().execute(page, context)
            return result

但简单的 try/catch 不够。需要考虑:

  • 幂等性:脚本执行到一半失败时,AI 是重新执行还是从失败点继续?
  • 状态传递:脚本已经在页面上输入了一半的表单,AI 接管时怎么知道已经填了什么?
  • 回退条件:什么时候 AI 处理完后交回给脚本,什么时候永久切换到 AI?

状态感知桥接

解决状态传递问题:每次操作前记录页面状态快照,AI 接管时从快照恢复上下文。

class StateAwareBridge:
    async def execute_step(self, step_name, script_fn, ai_fn, page):
        state = await self.capture_state(page)
 
        try:
            return await script_fn(page)
        except ScriptError as e:
            # 恢复到操作前的状态,交给 AI
            await self.restore_state(page, state)
            return await ai_fn(page, str(e))
 
    async def capture_state(self, page):
        return {
            "url": page.url,
            "title": await page.title(),
            "cookies": await page.context.cookies(),
            "visible_text": await page.text_content("body"),
        }

这样 AI 接管时知道:

  • 当前在哪个页面(url)
  • 已经操作到什么程度(visible_text)
  • 会话状态(cookies)

降级策略与熔断机制

如果脚本连续失败,说明页面结构可能已经发生变化。这种情况下每次都 try→AI 的效率很低。应该熔断。

class CircuitBreaker:
    def __init__(self, threshold=3, reset_after=300):
        self.failures = 0
        self.threshold = threshold
        self.reset_after = reset_after
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open
 
    async def execute(self, script_fn, ai_fn):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_after:
                self.state = "half-open"
            else:
                return await ai_fn()  # 直接走 AI
 
        try:
            result = await script_fn()
            self.failures = 0
            self.state = "closed"
            return result
        except ScriptError:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.threshold:
                self.state = "open"
                logger.info("Circuit opened: switching to AI permanently")
            return await ai_fn()

当脚本连续失败 3 次,熔断器打开,后续操作直接走 AI 不再尝试脚本。300 秒后半开,给脚本一次恢复机会。如果恢复失败,继续保持 AI 模式。

桥接层和数据流程

加载图表中...

监控指标

桥接层需要暴露的关键指标:

# 需要监控的指标
{
    "script_execution_count": 9500,      # 脚本执行次数
    "ai_execution_count": 500,           # AI 执行次数
    "fallback_rate": 0.05,               # 降级率(5%)
    "circuit_breaker_state": "closed",   # 熔断器状态
    "avg_script_time_ms": 120,           # 脚本平均耗时
    "avg_ai_time_ms": 3500,              # AI 平均耗时
    "script_error_types": {              # 脚本错误分布
        "TimeoutError": 300,
        "SelectorNotFoundError": 180,
        "NetworkError": 20,
    }
}

如果降级率超过 10%,说明选择器可能大面积过期,需要更新基线。

总结

桥接层是混合架构落地的关键。它解决三个问题:

  1. 统一接口——调用方不关心背后是脚本还是 AI
  2. 自动切换——脚本失败时透明切换到 AI,不中断业务
  3. 熔断保护——连续失败时停用脚本,避免无效重试

下一篇文章讨论 AI 也失败时怎么办:AI 失败回退到脚本的灰度兜底策略。

需要企业代理方案?

我们可根据目标站点、并发规模与稳定性目标提供定制方案。