Xây Dựng Vòng Lặp Agent Cho Phone Farm Android với ADB & UiAutomator2

Tại Sao Phone Farm Cần Agent Loop

Chạy tác vụ thủ công trên 10–50 thiết bị Android là thảm họa. App crash, session hết hạn, CAPTCHA xuất hiện. Bạn cần một agent loop bền vững cho mỗi thiết bị.

Kiến Trúc

Task Scheduler → Device Manager → Agent Loop (UiAutomator2 + OCR) → scrcpy monitor

1. Phát Hiện Thiết Bị

def list_devices():
    result = subprocess.run(["adb", "devices", "-l"], capture_output=True, text=True)
    devices = []
    for line in result.stdout.strip().splitlines()[1:]:
        if not line.strip(): continue
        parts = line.split()
        serial, state = parts[0], parts[1]
        devices.append(Device(serial=serial, state=state))
    return [d for d in devices if d.state == "device"]

2. Vòng Lặp Agent

async def agent_loop(serial: str, task_queue: asyncio.Queue):
    d = u2.connect(serial)
    while True:
        task = await task_queue.get()
        try:
            await execute_task(d, task)
        except Exception:
            await handle_recovery(d, serial)
        finally:
            task_queue.task_done()

3. OCR Fallback

async def ocr_tap(d, target_text: str):
    img = d.screenshot(format="opencv")
    result = ocr.ocr(img, cls=True)
    for line in result[0]:
        bbox, (text, conf) = line
        if target_text.lower() in text.lower() and conf > 0.8:
            cx = int((bbox[0][0] + bbox[2][0]) / 2)
            cy = int((bbox[0][1] + bbox[2][1]) / 2)
            d.click(cx, cy)
            return

4. Phục Hồi

async def handle_recovery(d, serial):
    for fn in [lambda: d.press("back"), lambda: d.press("home"), lambda: d.reboot()]:
        try:
            fn()
            await asyncio.sleep(2)
            if is_responsive(d): return
        except: continue

Kết Luận

• Một agent loop per thiết bị
• OCR fallback cho canvas UI
• Xây failure path trước happy path
• scrcpy tách biệt khỏi kênh automation