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%,说明选择器可能大面积过期,需要更新基线。
总结
桥接层是混合架构落地的关键。它解决三个问题:
- 统一接口——调用方不关心背后是脚本还是 AI
- 自动切换——脚本失败时透明切换到 AI,不中断业务
- 熔断保护——连续失败时停用脚本,避免无效重试
下一篇文章讨论 AI 也失败时怎么办:AI 失败回退到脚本的灰度兜底策略。
需要企业代理方案?
我们可根据目标站点、并发规模与稳定性目标提供定制方案。