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 неделя

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