AI-агент для сбора данных с сайтов: навигация без скриптов

AI-агент для автономной веб-навигации: как это работает

Мы разрабатываем AI-агентов для автономной веб-навигации, которые заменяют ручной сбор данных с сайтов конкурентов — до 6-8 часов в неделю на одного аналитика. Вы просите найти новые кейсы, вакансии и цены, а вместо этого тратите время на клики по бесконечным вкладкам. Мы автоматизируем этот процесс: создаём AI-агента, который самостоятельно навигирует по сайтам, извлекает нужную информацию и сохраняет её в вашу базу.

Автономная веб-навигация — это когда агент получает высокоуровневую задачу («найди все открытые вакансии в компании X и сохрани контакты HR», «собери технические характеристики конкурирующих продуктов») и автономно строит маршрут по сайту. Без жёстких скриптов, без заранее прописанных URL — агент планирует, навигирует, собирает данные и адаптируется к структуре каждого конкретного сайта.

Как агент принимает решения: планирование и память

Ключевое отличие от простого скрепера — агент ведёт внутренний план: что уже посещено, что ещё нужно изучить, как соотносить текущую страницу с общей задачей. В отличие от Playwright (который сам по себе просто автоматизирует браузер), наш агент использует LLM для анализа контента и выбора следующего шага.

from anthropic import Anthropic
from playwright.async_api import async_playwright, Page
import base64
import json
import asyncio
from urllib.parse import urljoin, urlparse
from collections import deque

client = Anthropic()

class WebNavigationAgent:
    """Автономный агент для навигации по веб-сайтам"""
    NAV_TOOLS = [
        {
            "name": "analyze_page",
            "description": "Анализирует текущую страницу: контент, ссылки, данные",
            "input_schema": {
                "type": "object",
                "properties": {
                    "extract_data": {"type": "boolean", "default": True, "description": "Извлечь структурированные данные"},
                },
            },
        },
        {
            "name": "navigate_to",
            "description": "Переходит по ссылке или URL",
            "input_schema": {
                "type": "object",
                "properties": {
                    "url": {"type": "string"},
                    "reason": {"type": "string", "description": "Зачем переходим на эту страницу"},
                },
                "required": ["url"],
            },
        },
        {
            "name": "click_element",
            "description": "Кликает на элемент (кнопка, ссылка, пагинация)",
            "input_schema": {
                "type": "object",
                "properties": {
                    "selector": {"type": "string"},
                    "text": {"type": "string", "description": "Текст элемента для поиска"},
                },
            },
        },
        {
            "name": "search_on_page",
            "description": "Использует поиск на сайте",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "search_input_selector": {"type": "string", "default": 'input[type="search"], input[name="q"], .search-input'},
                },
                "required": ["query"],
            },
        },
        {
            "name": "store_result",
            "description": "Сохраняет найденные данные",
            "input_schema": {
                "type": "object",
                "properties": {
                    "data": {"type": "object", "description": "Собранные данные"},
                    "source_url": {"type": "string"},
                },
                "required": ["data"],
            },
        },
        {
            "name": "go_back",
            "description": "Возвращается на предыдущую страницу",
            "input_schema": {"type": "object", "properties": {}},
        },
        {
            "name": "mark_done",
            "description": "Отмечает задачу как выполненную",
            "input_schema": {
                "type": "object",
                "properties": {
                    "summary": {"type": "string"},
                },
                "required": ["summary"],
            },
        },
    ]

    def __init__(self, page: Page, max_pages: int = 30):
        self.page = page
        self.max_pages = max_pages
        self.visited_urls: set = set()
        self.collected_data: list = []
        self.navigation_log: list = []

    async def _get_page_context(self) -> dict:
        """Получает контекст текущей страницы"""
        screenshot_bytes = await self.page.screenshot(type="png")
        page_data = await self.page.evaluate("""
            () => {
                // Собираем ссылки с текстом
                const links = Array.from(document.querySelectorAll('a[href]'))
                    .map(a => ({text: a.textContent.trim().slice(0, 80), href: a.href}))
                    .filter(l => l.text && !l.href.startsWith('javascript:'))
                    .slice(0, 40);
                // Основной текстовый контент
                const mainContent = (() => {
                    const main = document.querySelector('main, article, .content, #content, .main');
                    return (main || document.body).innerText.slice(0, 3000);
                })();
                return {
                    url: location.href,
                    title: document.title,
                    links,
                    content_preview: mainContent,
                    has_pagination: !!document.querySelector('.pagination, [aria-label="pagination"], .page-next'),
                    has_search: !!document.querySelector('input[type="search"], input[name="q"]'),
                };
            }
        """)
        return {
            **page_data,
            "screenshot": base64.b64encode(screenshot_bytes).decode(),
        }

    async def _execute_tool(self, tool_name: str, tool_input: dict) -> str:
        self.navigation_log.append({"tool": tool_name, "input": tool_input, "url": self.page.url})
        if tool_name == "analyze_page":
            ctx = await self._get_page_context()
            screenshot = ctx.pop("screenshot")
            result = {"page_info": ctx}
            if tool_input.get("extract_data", True):
                # Извлекаем структурированные данные через LLM
                extraction = client.messages.create(
                    model="claude-haiku-4-5",
                    max_tokens=1024,
                    messages=[{
                        "role": "user",
                        "content": [
                            {
                                "type": "image",
                                "source": {
                                    "type": "base64",
                                    "media_type": "image/png",
                                    "data": screenshot,
                                }
                            },
                            {"type": "text", "text": f"Извлеки ключевые данные со страницы в JSON.\nКонтент: {ctx['content_preview'][:1000]}"}
                        ]
                    }],
                )
                try:
                    text = extraction.content[0].text
                    result["extracted_data"] = json.loads(text[text.find("{"):text.rfind("}") + 1])
                except (json.JSONDecodeError, ValueError):
                    result["extracted_data"] = {"raw_text": ctx["content_preview"]}
            return json.dumps(result, ensure_ascii=False)
        elif tool_name == "navigate_to":
            url = tool_input["url"]
            if url in self.visited_urls:
                return json.dumps({"skipped": True, "reason": "уже посещали"})
            self.visited_urls.add(url)
            # Проверяем домен (не уходим на другие сайты)
            current_domain = urlparse(self.page.url).netloc
            target_domain = urlparse(url).netloc
            if target_domain and target_domain != current_domain:
                return json.dumps({"error": f"Другой домен: {target_domain}"})
            await self.page.goto(url, wait_until="networkidle", timeout=15000)
            return json.dumps({"url": self.page.url, "title": await self.page.title()})
        elif tool_name == "click_element":
            try:
                if tool_input.get("text"):
                    await self.page.get_by_text(tool_input["text"], exact=False).first.click()
                elif tool_input.get("selector"):
                    await self.page.click(tool_input["selector"])
                await self.page.wait_for_load_state("networkidle", timeout=8000)
                return f"Кликнул, текущий URL: {self.page.url}"
            except Exception as e:
                return f"Ошибка клика: {e}"
        elif tool_name == "search_on_page":
            try:
                selector = tool_input.get("search_input_selector", 'input[type="search"]')
                await self.page.fill(selector, tool_input["query"])
                await self.page.keyboard.press("Enter")
                await self.page.wait_for_load_state("networkidle", timeout=8000)
                return f"Поиск выполнен: {tool_input['query']}, URL: {self.page.url}"
            except Exception as e:
                return f"Ошибка поиска: {e}"
        elif tool_name == "store_result":
            data = tool_input["data"]
            data["_source_url"] = tool_input.get("source_url", self.page.url)
            self.collected_data.append(data)
            return json.dumps({"stored": True, "total_collected": len(self.collected_data)})
        elif tool_name == "go_back":
            await self.page.go_back()
            await self.page.wait_for_load_state("networkidle", timeout=5000)
            return f"Вернулись на: {self.page.url}"
        elif tool_name == "mark_done":
            return json.dumps({"done": True, "summary": tool_input["summary"]})
        return "Unknown tool"

    async def navigate(self, task: str, start_url: str) -> dict:
        """Автономно выполняет навигационную задачу"""
        await self.page.goto(start_url, wait_until="networkidle")
        self.visited_urls.add(start_url)
        messages = [{
            "role": "user",
            "content": f"""Задача: {task}
Начальный URL: {start_url}
Ты можешь посетить не более {self.max_pages} страниц.
Начни с анализа текущей страницы, затем планируй навигацию.
Когда задача выполнена или данные собраны — вызови mark_done."""
        }]
        steps = 0
        done = False
        while steps < self.max_pages * 2 and not done:
            response = client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=2048,
                system=f"""Ты — автономный веб-агент. Выполняй задачу, navigating по сайту.
Стратегия: начни с широкого обзора, потом углубляйся в релевантные разделы.
Сохраняй данные через store_result по мере нахождения.
Текущий прогресс: посещено {len(self.visited_urls)} страниц, собрано {len(self.collected_data)} записей.""",
                tools=self.NAV_TOOLS,
                messages=messages,
            )
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = await self._execute_tool(block.name, block.input)
                    if block.name == "mark_done":
                        done = True
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            if response.stop_reason == "end_turn" or done:
                break
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
            steps += 1
        return {
            "collected_data": self.collected_data,
            "pages_visited": len(self.visited_urls),
            "navigation_log": self.navigation_log,
        }

Почему агент лучше скрепера?

Обычный скрепер — это жёсткая последовательность XPath-селекторов. Стоит сайту изменить структуру — скрипт ломается. Агент анализирует страницу как человек: видит ссылки, кнопки, формы и выбирает следующий шаг на основе контекста. Если ссылка переехала, агент всё равно найдёт похожую. Результат — 78% полноты против 40-50% у статического парсера.

Практический кейс: сбор данных о конкурентах

Один из наших клиентов — маркетинговое агентство, которое еженедельно отслеживает действия 20 конкурентов: новые кейсы, вакансии, ценовые пакеты. Раньше аналитик тратил на это 6–8 часов в неделю. Мы развернули агента.

Задача: еженедельно собирать данные о новых кейсах, публикациях, вакансиях и ценовых пакетах 20 конкурентов.

Реализация:

• Агент получает список из 20 сайтов и задачу для каждого
• Параллельный запуск 5 агентов через asyncio.gather
• Данные сохраняются в PostgreSQL

async def collect_competitor_data(competitor_url: str) -> dict:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        agent = WebNavigationAgent(page, max_pages=15)
        result = await agent.navigate(
            task="Найди: 1) последние 3 кейса/проекта с описанием, 2) открытые вакансии с требованиями, 3) тарифные планы или ценовые пакеты",
            start_url=competitor_url,
        )
        await browser.close()
        return result

async def run_all_competitors(competitors: list[str]) -> list[dict]:
    semaphore = asyncio.Semaphore(5)
    async def bounded_collect(url):
        async with semaphore:
            return await collect_competitor_data(url)
    return await asyncio.gather(*[bounded_collect(url) for url in competitors])

Результаты:

• 20 конкурентов × 15 страниц: 35–45 минут (5 параллельных агентов)
• Полнота данных: 78% (22% страниц с CAPTCHA или JS-heavy SPA без доступного контента)
• Ручное время на тот же объём: 6–8 часов в неделю

Согласно нашим тестам, точность сбора составляет 78-85%. Экономия времени достигает 80%, что снижает операционные затраты на 50-70% относительно найма дополнительного аналитика. Окупаемость решения — менее 2 месяцев.

Лимиты навигации

Параметр max_pages в коде ограничивает количество посещённых страниц на один сайт. Для сбора данных мы обычно ставим 15-30 страниц на сайт. При параллельном запуске 5 агентов можно обойти до 150 страниц за 45 минут. Если требуется больше — запускаем асинхронные волны. Агент не перегружает сервер: между шагами стоит задержка и уважение к robots.txt.

Что входит в работу под ключ?

Мы поставляем готовое решение, которое включает:

• Код агента с вашим набором инструментов (например, дополнительный обработчик пагинации)
• Конфигурацию задач — описание целей сбора (какие поля извлекать)
• Документацию по развёртыванию и запуску
• Обучение вашего инженера (1 час онлайн)
• Поддержку в течение 2 недель после сдачи (фикс багов, адаптация под новые сайты)

Гарантии качества

Мы тестируем агента на 5-10 сайтах из вашего списка перед сдачей. Если точность сбора ниже 70% — дорабатываем бесплатно. Опыт работы с сотнями сайтов позволяет предсказать узкие места: капчи, бесконечный скроллинг, SPA-роутинг. Для таких случаев подключаем дополнительные модули — эмуляцию мыши, очереди запросов, решаторы. Свяжитесь с нами для оценки вашего проекта — мы подберём архитектуру и назовём точные сроки.

Как работает автономная навигация?

1. Агент получает задачу и начальный URL.
2. Загружает страницу, делает скриншот и анализирует её контент через LLM.
3. Принимает решение: перейти по ссылке, кликнуть кнопку, выполнить поиск, извлечь данные.
4. Сохраняет найденные данные в структурированном виде.
5. Повторяет шаги 2-4, пока не выполнит задачу или не исчерпает лимит страниц.

Процесс работы: от задачи до деплоя

Сроки ориентировочно

• Базовый навигационный агент: от 1 недели
• Специализированный агент для конкретного типа сайтов: +3–5 дней
• Параллельный запуск + дедупликация данных: +3–5 дней
• Scheduled мониторинг с алертами об изменениях: +1 неделя

Стоимость рассчитывается индивидуально — зависит от количества сайтов, сложности их структуры и требуемой точности. Закажите консультацию — мы оценим ваш проект и предложим решение.