ניווט אוטונומי באתרי אינטרנט עם סוכן AI

סוכן בינה מלאכותית לניווט אוטונומי באינטרנט: איך זה עובד

אתה מבלה 6-8 שעות בשבוע באיסוף נתונים ידני מאתרי מתחרים — לוחץ על טאבים אינסופיים, מעתיק מחירים ומדביק לתוך גיליונות אלקטרוניים. אנחנו הופכים את התהליך הזה לאוטומטי: אנו בונים סוכן בינה מלאכותית שמנווט באתרים באופן אוטונומי, מחלץ את המידע שאתה צריך, ומאחסן אותו ישירות לתוך מסד הנתונים שלך.

ניווט אוטונומי באינטרנט פירושו שהסוכן מקבל משימה ברמה גבוהה ("מצא את כל המשרות הפתוחות בחברה X ושמור פרטי קשר של משאבי אנוש", "אסוף מפרטים טכניים של מוצרים מתחרים") ומתכנן באופן אוטונומי מסלול דרך האתר. ללא סקריפטים נוקשים, ללא כתובות 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 סוכנים באמצעות 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, }
• הנתונים נשמרים ב-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 או SPA כבד ב-JS ללא תוכן נגיש)
• זמן ידני לאותה כמות: 6-8 שעות בשבוע

בהתבסס על הבדיקות שלנו, דיוק האיסוף הוא 78-85%. חיסכון בזמן מגיע ל-80%, ומפחית עלויות תפעול ב-50-70% בהשוואה להעסקת אנליסט נוסף. תקופת החזר השקעה: פחות מחודשיים.

מגבלות ניווט

הפרמטר asyncio.gather בקוד מגביל את מספר העמודים שנצפים בכל אתר. עבור איסוף נתונים אנחנו בדרך כלל מגדירים 15-30 עמודים לכל אתר. עם 5 סוכנים מקביליים, אנחנו יכולים לכסות עד 150 עמודים ב-45 דקות. אם יש צורך ביותר, אנחנו מפעילים גלים אסינכרוניים. הסוכן לא מעמיס על השרת: יש השהיה בין הצעדים וכיבוד של 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]) .

מה כלול בפתרון סוהר?

אנחנו מספקים פתרון מוכן שכולל:

• קוד הסוכן עם ערכת הכלים המותאמת אישית שלך (לדוגמה, מטפל פאג'ינציה נוסף)
• הגדרת משימה — תיאור מטרות האיסוף (אילו שדות לחלץ)
• תיעוד לפריסה והפעלה
• הדרכה למהנדס שלך (שעה אחת אונליין)
• תמיכה למשך שבועיים לאחר המסירה (תיקוני באגים, התאמה לאתרים חדשים)

הבטחות איכות

אנחנו בודקים את הסוכן על 5-10 אתרים מהרשימה שלך לפני המסירה. אם דיוק האיסוף נמוך מ-70%, אנחנו מתקנים אותו בחינם. ניסיון עם מאות אתרים מאפשר לנו לחזות צווארי בקבוק: CAPTCHA, גלילה אינסופית, ניתוב SPA. למקרים כאלה אנחנו מוסיפים מודולים נוספים — אמולציית עכבר, תורי בקשות, פותרי CAPTCHA. צור קשר להערכת פרויקט — נבחר את הארכיטקטורה ונספק לוחות זמנים מדויקים.

איך ניווט אוטונומי עובד

1. הסוכן מקבל משימה וכתובת URL התחלתית.
2. הוא טוען את העמוד, מצלם מסך ומנתח תוכן באמצעות LLM.
3. הוא מחליט: לעקוב אחרי קישור, ללחוץ על כפתור, לבצע חיפוש, לחלץ נתונים.
4. הוא שומר את הנתונים שנמצאו בצורה מובנית.
5. הוא חוזר על שלבים 2-4 עד שהמשימה הושלמה או שמגיעים למגבלת העמודים.

תהליך: ממשימה לפריסה

לוחות זמנים משוערים

• סוכן ניווט בסיסי: החל משבוע אחד
• סוכן מתמחה לסוג אתר ספציפי: +3-5 ימים
• הפעלה מקבילית + הסרת כפילויות נתונים: +3-5 ימים
• ניטור מתוזמן עם התראות שינויים: +שבוע אחד

העלות מחושבת באופן אישי — תלויה במספר האתרים, במורכבות המבנית שלהם ובדיוק הנדרש. הזמינו ייעוץ — נעריך את הפרויקט שלך ונציע פתרון.