Playwright 到 AI 智能体迁移（二）：双模式 API 设计与灰度切换
同一套接口背后可以走 AI 也可以走脚本。配置开关切换、A/B 对比、灰度递增流量——迁移不需要停机切换。
亿牛云技术团队2026年5月8日1 分钟阅读
迁移不需要停机
传统系统迁移的问题是：一次性切换，出错了回滚代价高。
AI 浏览器自动化的迁移不需要这样。因为脚本和 AI 智能体操作的是同一个浏览器 API（CDP），可以在同一套接口后面并行运行。
双模式接口设计
class ExtractionEngine:
"""同一套接口，两种实现"""
def __init__(self):
self.script_engine = ScriptEngine()
self.ai_engine = AIEngine()
async def extract_product_info(self, url, fields, mode="auto"):
"""
提取产品信息
mode: "script", "ai", "auto", "compare"
"""
if mode == "script":
return await self.script_engine.extract(url, fields)
elif mode == "ai":
return await self.ai_engine.extract(url, fields)
elif mode == "compare":
# 对比模式——同时跑两种实现
script_result, ai_result = await asyncio.gather(
self.script_engine.extract(url, fields),
self.ai_engine.extract(url, fields),
)
return self.compare_results(script_result, ai_result)
else:
# auto 模式：根据配置决定
if self.should_use_ai(url):
return await self.ai_engine.extract(url, fields)
return await self.script_engine.extract(url, fields)
def should_use_ai(self, url):
"""动态决策用哪种实现"""
# 最近脚本失败次数
recent_failures = self.get_recent_failures(url)
# 如果最近 10 次中失败超过 3 次，切换到 AI
return recent_failures >= 3调用方不需要关心是脚本还是 AI 在执行——接口是一致的。
灰度切换方案
class GradualMigration:
def __init__(self, module_name):
self.module_name = module_name
# 配置：{"script": 0.9, "ai": 0.1} → 10% 流量走 AI
self.traffic_split = {"script": 1.0, "ai": 0.0}
self.metrics = {
"script_success": 0,
"ai_success": 0,
"script_duration": [],
"ai_duration": [],
}
def route(self, task):
"""根据分流比例决定走哪条路径"""
if random.random() < self.traffic_split["ai"]:
return "ai"
return "script"
async def execute_and_compare(self, task):
"""对比模式——两个路径都走，比较结果"""
script_task = asyncio.create_task(self.script_engine(task))
ai_task = asyncio.create_task(self.ai_engine(task))
script_result, ai_result = await asyncio.gather(
script_task, ai_task, return_exceptions=True
)
# 记录比较结果
comparison = {
"task": task.id,
"script_success": not isinstance(script_result, Exception),
"ai_success": not isinstance(ai_result, Exception),
"results_match": self.results_match(script_result, ai_result),
"script_duration": script_result.duration if hasattr(script_result, 'duration') else 0,
"ai_duration": ai_result.duration if hasattr(ai_result, 'duration') else 0,
}
await self.store_comparison(comparison)
# 默认返回脚本结果（稳定的实现）
if not isinstance(script_result, Exception):
return script_result
return ai_result递增流量控制
class TrafficController:
def __init__(self, module_name):
self.module_name = module_name
self.current_ai_percent = 0
self.target_percent = 0
def set_ai_percent(self, percent):
"""设置 AI 流量百分比"""
self.target_percent = percent
self.transition_start = time.time()
def get_current_ai_percent(self):
"""平滑递增——每次增加 5%"""
target = self.target_percent
current = self.current_ai_percent
if current < target:
increment = min(5, target - current)
self.current_ai_percent = current + increment
return self.current_ai_percent灰度递增流程：
第 1 天：AI 流量 = 1% → 观察：是否报错？
第 2 天：AI 流量 = 5% → 观察：成功率是否下降？
第 3 天：AI 流量 = 10% → 观察：延迟是否增加？
第 4-7 天：每日增加 10-20%
第 8 天：AI 流量 = 100%每一步都设置自动回滚条件。成功率下降超过 2% 自动回退。
自动回滚条件
class AutoRollback:
def __init__(self):
self.baseline = {
"success_rate": 0.98,
"avg_duration_ms": 2000,
"error_rate": 0.01,
}
def should_rollback(self, metrics):
reasons = []
if metrics["success_rate"] < self.baseline["success_rate"] - 0.02:
reasons.append(f"Success rate dropped: "
f"{metrics['success_rate']:.1%} < "
f"{self.baseline['success_rate']:.1%}")
if metrics["avg_duration_ms"] > self.baseline["avg_duration_ms"] * 2:
reasons.append(f"Duration doubled: "
f"{metrics['avg_duration_ms']}ms > "
f"{self.baseline['avg_duration_ms']}ms")
return reasons总结
从 Playwright 到 AI 智能体的迁移不需要停机切换。核心设计：
- 双模式接口——同一套调用接口，背后可走脚本或 AI
- 对比模式——同时运行两种实现，验证结果一致性
- 灰度递增——从 1% 开始逐步扩大 AI 流量
- 自动回滚——成功率下降或延迟增加时自动回退
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。