AI-агент для автоматизации браузера (Web Agent)

AI-агент для автоматизации браузера и веб-навигации

Разница между RPA-ботом и AI веб-агентом: RPA ломается при изменении верстки, AI-агент понимает семантику страницы. Вместо click("#submit-btn-v2") агент ищет кнопку с текстом «Оформить заказ» и находит её даже после редизайна.

Архитектура: Playwright + LLM

from playwright.async_api import async_playwright, Page
from anthropic import Anthropic
import asyncio
import base64
import json

client = Anthropic()

class AIWebAgent:

    def __init__(self, headless: bool = True):
        self.headless = headless

    async def run(self, task: str, start_url: str) -> str:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=self.headless)
            context = await browser.new_context(
                viewport={"width": 1280, "height": 800},
                user_agent="Mozilla/5.0 (compatible; research-bot/1.0)"
            )
            page = await context.new_page()
            await page.goto(start_url)
            result = await self._agent_loop(page, task)
            await browser.close()
            return result

    async def _get_page_state(self, page: Page) -> dict:
        screenshot = await page.screenshot(type="png")
        screenshot_b64 = base64.standard_b64encode(screenshot).decode()

        elements = await page.evaluate("""() => {
            const interactive = document.querySelectorAll(
                'a, button, input, select, textarea, [role="button"]'
            );
            return Array.from(interactive).slice(0, 50).map((el, idx) => ({
                index: idx,
                tag: el.tagName.toLowerCase(),
                text: el.textContent?.trim().slice(0, 80) || '',
                type: el.type || '',
                placeholder: el.placeholder || '',
            }));
        }""")

        return {
            "url": page.url,
            "title": await page.title(),
            "screenshot": screenshot_b64,
            "elements": elements,
        }

    async def _execute_action(self, page: Page, action: dict) -> str:
        action_type = action.get("type")
        try:
            if action_type == "click":
                if "text" in action:
                    await page.get_by_text(action["text"]).first.click()
                elif "element_index" in action:
                    elements = await page.query_selector_all(
                        'a, button, input, select, textarea, [role="button"]'
                    )
                    if action["element_index"] < len(elements):
                        await elements[action["element_index"]].click()
            elif action_type == "fill":
                await page.fill(action["selector"], action["value"])
            elif action_type == "navigate":
                await page.goto(action["url"])
            elif action_type == "scroll":
                amount = action.get("amount", 500)
                direction = 1 if action.get("direction", "down") == "down" else -1
                await page.evaluate(f"window.scrollBy(0, {direction * amount})")
            elif action_type == "extract":
                return await page.evaluate(
                    f'document.querySelector({json.dumps(action.get("selector", "body"))})?.textContent?.trim()'
                )
            await asyncio.sleep(0.5)
            return "success"
        except Exception as e:
            return f"error: {e}"

    async def _agent_loop(self, page: Page, task: str) -> str:
        messages = []
        system = """Ты — AI веб-агент. Выполняй задачи в браузере.

Доступные действия (верни JSON):
- {"type": "click", "text": "текст кнопки"}
- {"type": "fill", "selector": "CSS", "value": "текст"}
- {"type": "navigate", "url": "https://..."}
- {"type": "scroll", "direction": "down", "amount": 500}
- {"type": "extract", "selector": ".result"}
- {"type": "done", "result": "итоговый результат"}

Формат ответа:
REASONING: <что видишь и что планируешь>
ACTION: <JSON с одним действием>"""

        for _ in range(20):
            state = await self._get_page_state(page)
            user_content = [
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": state["screenshot"]}},
                {"type": "text", "text": f"Задача: {task}\nURL: {state['url']}\nЭлементы:\n{json.dumps(state['elements'][:20], ensure_ascii=False)}"}
            ]
            messages.append({"role": "user", "content": user_content})

            response = client.messages.create(
                model="claude-opus-4-5",
                max_tokens=1024,
                system=system,
                messages=messages,
            )
            reply = response.content[0].text
            messages.append({"role": "assistant", "content": reply})

            try:
                action_line = [l for l in reply.split("\n") if l.startswith("ACTION:")][0]
                action = json.loads(action_line.replace("ACTION:", "").strip())
            except (IndexError, json.JSONDecodeError):
                continue

            if action.get("type") == "done":
                return action.get("result", "Task completed")

            await self._execute_action(page, action)

        return "Max iterations reached"

Конкурентный мониторинг цен

async def monitor_competitor_prices(products: list[str], sites: list[str]) -> dict:
    agent = AIWebAgent(headless=True)
    results = {}

    for product in products:
        task = f"Найди цену на '{product}', верни только число в рублях"
        for site in sites:
            price = await agent.run(task, f"https://{site}/search/?text={product}")
            results.setdefault(product, {})[site] = price

    return results

# Использование
prices = asyncio.run(monitor_competitor_prices(
    products=["Цемент М500 50кг", "Арматура 12мм 6м"],
    sites=["leroymerlin.ru", "obi.ru", "220-volt.ru"]
))

Обработка anti-bot защиты

async def setup_stealth_context(playwright):
    browser = await playwright.chromium.launch(
        headless=False,
        args=["--disable-blink-features=AutomationControlled"]
    )
    context = await browser.new_context(
        viewport={"width": 1366, "height": 768},
        locale="ru-RU",
        timezone_id="Europe/Moscow",
    )
    await context.add_init_script(
        "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
    )
    return context

Типовые применения

Мониторинг конкурентов — цены, акции, ассортимент на нескольких площадках автоматически каждый день.

Заполнение форм — подача заявок на тендеры, регистрация на мероприятия, заполнение государственных порталов.

Сбор отзывов — агрегация оценок и отзывов с нескольких площадок для анализа репутации.

Регрессионное тестирование UI — агент проходит пользовательские сценарии и фиксирует визуальные изменения.

Кейс: сеть строительных магазинов

Задача: ежедневный мониторинг цен 200 SKU у 5 конкурентов. Ручной процесс занимал 4 часа в день у менеджера по закупкам.

AI-агент обходит сайты конкурентов, извлекает актуальные цены и наличие, формирует Excel-отчёт с отклонениями. Задача выполняется за 35 минут в автоматическом режиме.

Результат: менеджер тратит 15 минут на анализ готового отчёта вместо 4 часов сбора данных. Реакция на изменение цен ускорилась с 1-2 дней до нескольких часов.

Сроки

• Базовый агент для конкретного сайта: 3–5 дней
• Универсальный агент с визуальным восприятием: 1–2 недели
• Мониторинг цен с отчётами: 1 неделя
• Anti-bot защита, ротация прокси: +1 неделя