AI-агент для автономной навигации по веб-сайтам
Автономная веб-навигация — это когда агент получает высокоуровневую задачу («найди все открытые вакансии в компании X и сохрани контакты HR», «собери технические характеристики конкурирующих продуктов») и самостоятельно строит маршрут по сайту. Без жёстких скриптов, без заранее прописанных URL — агент планирует, навигирует, собирает данные и адаптируется к структуре каждого конкретного сайта.
Агент с планированием и памятью
Ключевое отличие от простого скрепера — агент ведёт внутренний план: что уже посещено, что ещё нужно изучить, как соотносить текущую страницу с общей задачей.
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,
}
Практический кейс: сбор данных о конкурентах
Задача: компания хотела еженедельно собирать данные о новых кейсах, публикациях, вакансиях и ценовых пакетах 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 часов в неделю
Сроки
- Базовый навигационный агент: 1 неделя
- Специализированный агент для конкретного типа сайтов: +3–5 дней
- Параллельный запуск + дедупликация данных: +3–5 дней
- Scheduled мониторинг с алертами об изменениях: +1 неделя