Playwright + AI Agent Hybrid Architecture (Part 2): A Bridge Layer for Scripts and AI
Same task — scripts handle navigation and extraction, AI handles exceptions and adaptation. The bridge layer coordinates both with transparent switching.
The Core Problem
You have a task: "login to e-commerce site → search products → extract prices → submit to comparison system." Search and extraction are structurally stable — use scripts. Login may trigger CAPTCHA. Extraction pages may change.
The same task flow needs scripts for certain steps and AI for others — with transparent switching when errors occur.
Unified Interface
The core of a bridge layer is a unified operation interface:
class HybridExecutor:
async def execute_with_fallback(self, operation, page, context):
try:
return await ScriptOperation().execute(page, context)
except (TimeoutError, SelectorNotFoundError, NetworkError) as e:
logger.info(f"Script failed, falling back to AI: {str(e)}")
return await AIOperation().execute(page, context)But try/catch alone isn't enough. Consider:
- Idempotency: Script failed mid-execution — does AI restart or continue?
- State passing: Script already filled half a form — how does AI know what's been filled?
- Fallback conditions: When does AI hand back to scripts? When does it stay permanent?
State-Aware Bridge
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:
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"),
}When AI takes over, it knows:
- Current page (url)
- Progress so far (visible_text)
- Session state (cookies)
Circuit Breaker
If scripts fail repeatedly, page structure may have changed. Every try→AI cycle is wasteful. Trip a circuit breaker.
class CircuitBreaker:
def __init__(self, threshold=3, reset_after=300):
self.failures = 0
self.threshold = threshold
self.reset_after = reset_after
self.state = "closed"
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()
try:
result = await script_fn()
self.failures = 0
self.state = "closed"
return result
except ScriptError:
self.failures += 1
if self.failures >= self.threshold:
self.state = "open"
return await ai_fn()3 consecutive failures → circuit opens → direct AI → half-open after 300s for recovery attempt.
Data Flow
加载图表中...
Key Metrics
{
"script_execution_count": 9500,
"ai_execution_count": 500,
"fallback_rate": 0.05,
"circuit_breaker_state": "closed",
"avg_script_time_ms": 120,
"avg_ai_time_ms": 3500,
}A fallback rate > 10% indicates selectors need updating.
Summary
The bridge layer solves three problems:
- Unified interface — caller doesn't know if it's script or AI
- Automatic switching — transparent fallback on failure
- Circuit protection — disable scripts on consecutive failure, avoid wasted retries
Next: what happens when AI also fails — graceful degradation strategies.
Need an enterprise proxy plan?
We can tailor architecture to your target domains, concurrency, and reliability goals.