Локальные LLM на практике: Строим мультимодальный краулер событий с нуля

Проблема: Почему облачные AI не подходят для краулинга событий?

Представьте: вам нужно ежедневно мониторить сотни сайтов с расписаниями мероприятий, концертов, выставок. Классические парсеры справляются с текстом, но как быть с картинками, баннерами, графиками расписаний? Облачные AI вроде GPT-4V удобны, но:

• Дорого — каждый анализ изображения стоит денег
• Медленно — задержки API при массовой обработке
• Не приватно — ваши данные уходят в облако
• Нет контроля — ограничения API, квоты, баны

Архитектура решения: как работает наш краулер

Наш мультимодальный краулер состоит из трёх ключевых компонентов:

1. Веб-сканер — собирает HTML и изображения с целевых страниц
2. Мультимодальный анализатор — GLM-4.6V анализирует и текст, и визуальный контент
3. Оркестратор — управляет пайплайном, сохраняет результаты в базу

Пошаговый план реализации

1 Подготовка среды и установка GLM-4.6V

Сначала настроим Ollama — самый простой способ запускать локальные LLM. Если вы сомневаетесь в выборе инструмента, сравнение LM Studio и llama.cpp поможет сделать правильный выбор.

# Установка Ollama (Linux/macOS)
curl -fsSL https://ollama.ai/install.sh | sh

# Скачивание GLM-4.6V (требуется ~8GB VRAM для полной версии)
ollama pull glm4v

# Для слабых систем используем квантованную версию
ollama pull glm4v:4bit  # Требуется ~4GB VRAM

2 Создание базового краулера на Playwright

Playwright идеально подходит для нашего случая: он умеет работать с JavaScript-сайтами и делать скриншоты.

# requirements.txt
playwright
beautifulsoup4
requests
pillow
chromadb
sqlalchemy
ollama

# crawler.py
import asyncio
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup
import os
from datetime import datetime

class EventCrawler:
    def __init__(self, output_dir="data"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        os.makedirs(f"{output_dir}/screenshots", exist_ok=True)
    
    async def crawl_page(self, url):
        """Основной метод краулинга страницы"""
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            context = await browser.new_context(viewport={'width': 1920, 'height': 1080})
            page = await context.new_page()
            
            try:
                await page.goto(url, wait_until="networkidle")
                
                # Сохраняем HTML
                html = await page.content()
                
                # Делаем полный скриншот
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                screenshot_path = f"{self.output_dir}/screenshots/{timestamp}.png"
                await page.screenshot(path=screenshot_path, full_page=True)
                
                # Извлекаем текст
                soup = BeautifulSoup(html, 'html.parser')
                text_content = soup.get_text(separator='\n', strip=True)
                
                # Находим все изображения на странице
                images = []
                for img in soup.find_all('img'):
                    src = img.get('src', '')
                    if src and not src.startswith('data:'):
                        images.append(src)
                
                return {
                    'url': url,
                    'html': html,
                    'text': text_content,
                    'screenshot': screenshot_path,
                    'images': images,
                    'timestamp': timestamp
                }
                
            finally:
                await browser.close()

# Пример использования
async def main():
    crawler = EventCrawler()
    events_sites = [
        "https://www.afisha.ru/",
        "https://www.timepad.ru/"
    ]
    
    for url in events_sites:
        data = await crawler.crawl_page(url)
        print(f"Собрано с {url}: {len(data['text'])} символов, {len(data['images'])} изображений")

if __name__ == "__main__":
    asyncio.run(main())

3 Мультимодальный анализатор на GLM-4.6V

Теперь самое интересное — анализ собранных данных с помощью локальной LLM. Мы будем использовать как текстовый, так и визуальный режимы модели.

# analyzer.py
import ollama
from PIL import Image
import base64
from io import BytesIO
import json

class MultimodalAnalyzer:
    def __init__(self, model="glm4v"):
        self.model = model
    
    def analyze_image(self, image_path, prompt):
        """Анализ изображения с помощью GLM-4.6V"""
        with open(image_path, "rb") as f:
            image_bytes = f.read()
        
        # Конвертируем в base64
        image_b64 = base64.b64encode(image_bytes).decode('utf-8')
        
        response = ollama.chat(
            model=self.model,
            messages=[{
                'role': 'user',
                'content': prompt,
                'images': [image_b64]
            }]
        )
        return response['message']['content']
    
    def analyze_text(self, text, prompt):
        """Анализ текста"""
        # Если текст очень длинный, разбиваем на части
        if len(text) > 8000:
            chunks = [text[i:i+8000] for i in range(0, len(text), 8000)]
            results = []
            for chunk in chunks:
                response = ollama.chat(
                    model=self.model,
                    messages=[{
                        'role': 'user',
                        'content': f"{prompt}\n\nТекст для анализа:\n{chunk}"
                    }]
                )
                results.append(response['message']['content'])
            return "\n\n".join(results)
        else:
            response = ollama.chat(
                model=self.model,
                messages=[{
                    'role': 'user',
                    'content': f"{prompt}\n\nТекст для анализа:\n{text}"
                }]
            )
            return response['message']['content']
    
    def extract_events_from_screenshot(self, screenshot_path):
        """Извлечение событий из скриншота"""
        prompt = """Проанализируй изображение веб-страницы и найди все упоминания событий, мероприятий, концертов, выставок.
        Для каждого события определи:
        1. Название события
        2. Дату и время (если указаны)
        3. Место проведения
        4. Краткое описание
        5. Цену (если указана)
        
        Верни результат в формате JSON."""
        
        result = self.analyze_image(screenshot_path, prompt)
        return self._parse_json_response(result)
    
    def extract_events_from_text(self, text):
        """Извлечение событий из текста"""
        prompt = """Извлеки из текста информацию о событиях, мероприятиях, концертах, выставках.
        Структурируй информацию по следующим полям:
        - title: название события
        - date: дата и время
        - location: место проведения
        - description: краткое описание
        - price: цена или "бесплатно"
        - category: категория (концерт, выставка, спектакль и т.д.)
        
        Верни результат в виде списка JSON объектов."""
        
        result = self.analyze_text(text, prompt)
        return self._parse_json_response(result)
    
    def _parse_json_response(self, text):
        """Пытаемся извлечь JSON из ответа модели"""
        try:
            # Ищем JSON в тексте
            import re
            json_match = re.search(r'\[.*\]|\{.*\}', text, re.DOTALL)
            if json_match:
                return json.loads(json_match.group())
            else:
                return {"error": "JSON не найден в ответе", "raw_response": text}
        except json.JSONDecodeError as e:
            return {"error": f"Ошибка парсинга JSON: {str(e)}", "raw_response": text}

# Пример использования
analyzer = MultimodalAnalyzer()

# Анализ скриншота
events_from_image = analyzer.extract_events_from_screenshot("data/screenshots/20241215_143000.png")
print(f"Найдено событий на изображении: {len(events_from_image) if isinstance(events_from_image, list) else 0}")

# Анализ текста
with open("data/page_text.txt", "r", encoding="utf-8") as f:
    text = f.read()
events_from_text = analyzer.extract_events_from_text(text)
print(f"Найдено событий в тексте: {len(events_from_text) if isinstance(events_from_text, list) else 0}")

4 Оркестрация и хранение данных

Создадим систему, которая координирует работу всех компонентов и сохраняет результаты.

# orchestrator.py
import sqlite3
import json
from datetime import datetime
import chromadb
from chromadb.config import Settings
import hashlib

class EventOrchestrator:
    def __init__(self, db_path="events.db"):
        self.db_path = db_path
        self._init_database()
        
        # Инициализация векторной БД для семантического поиска
        self.chroma_client = chromadb.Client(Settings(
            chroma_db_impl="duckdb+parquet",
            persist_directory="./chroma_db"
        ))
        self.collection = self.chroma_client.get_or_create_collection("events")
    
    def _init_database(self):
        """Создаем структуру БД"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            description TEXT,
            event_date TEXT,
            location TEXT,
            price TEXT,
            category TEXT,
            source_url TEXT,
            screenshot_path TEXT,
            extracted_from TEXT, -- 'text' или 'image'
            confidence REAL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS crawled_pages (
            url TEXT PRIMARY KEY,
            last_crawled TIMESTAMP,
            screenshot_path TEXT,
            status TEXT
        )
        ''')
        
        conn.commit()
        conn.close()
    
    def save_event(self, event_data, source_url, extracted_from, confidence=0.8):
        """Сохраняем событие в БД и векторное хранилище"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Генерируем ID для вектора
        event_id = hashlib.md5(
            f"{event_data.get('title', '')}{event_data.get('event_date', '')}".encode()
        ).hexdigest()
        
        # Сохраняем в SQLite
        cursor.execute('''
        INSERT OR REPLACE INTO events 
        (title, description, event_date, location, price, category, source_url, extracted_from, confidence)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (
            event_data.get('title', ''),
            event_data.get('description', ''),
            event_data.get('date', ''),
            event_data.get('location', ''),
            event_data.get('price', ''),
            event_data.get('category', ''),
            source_url,
            extracted_from,
            confidence
        ))
        
        # Сохраняем в ChromaDB для семантического поиска
        self.collection.add(
            documents=[json.dumps(event_data, ensure_ascii=False)],
            metadatas=[{
                "title": event_data.get('title', ''),
                "category": event_data.get('category', ''),
                "date": event_data.get('date', ''),
                "source": source_url
            }],
            ids=[event_id]
        )
        
        conn.commit()
        conn.close()
        
        return cursor.lastrowid
    
    def search_events(self, query, n_results=5):
        """Поиск событий по семантической схожести"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        
        events = []
        for doc, metadata in zip(results['documents'][0], results['metadatas'][0]):
            event_data = json.loads(doc)
            event_data['similarity_score'] = results['distances'][0][0] if results['distances'] else 0
            events.append(event_data)
        
        return events

# Основной пайплайн
class EventCrawlerPipeline:
    def __init__(self):
        self.crawler = EventCrawler()
        self.analyzer = MultimodalAnalyzer()
        self.orchestrator = EventOrchestrator()
    
    async def process_url(self, url):
        """Полный цикл обработки URL"""
        print(f"Начинаем обработку: {url}")
        
        # 1. Краулинг
        page_data = await self.crawler.crawl_page(url)
        print(f"Собрано данных: {len(page_data['text'])} символов")
        
        # 2. Анализ текста
        print("Анализируем текст...")
        events_from_text = self.analyzer.extract_events_from_text(page_data['text'])
        
        if isinstance(events_from_text, list):
            for event in events_from_text:
                self.orchestrator.save_event(event, url, "text", confidence=0.7)
            print(f"Сохранено событий из текста: {len(events_from_text)}")
        
        # 3. Анализ изображения
        print("Анализируем скриншот...")
        events_from_image = self.analyzer.extract_events_from_screenshot(page_data['screenshot'])
        
        if isinstance(events_from_image, list):
            for event in events_from_image:
                self.orchestrator.save_event(event, url, "image", confidence=0.6)
            print(f"Сохранено событий из изображения: {len(events_from_image)}")
        
        print(f"Обработка {url} завершена")
        
        return {
            "text_events": len(events_from_text) if isinstance(events_from_text, list) else 0,
            "image_events": len(events_from_image) if isinstance(events_from_image, list) else 0
        }

Запуск и мониторинг системы

Создадим основной скрипт для запуска краулера и простой веб-интерфейс для мониторинга.

# main.py
import asyncio
import schedule
import time
from orchestrator import EventCrawlerPipeline

# Список сайтов для мониторинга
TARGET_SITES = [
    "https://www.afisha.ru/",
    "https://www.timepad.ru/",
    "https://www.kassir.ru/",
    "https://peterburg2.ru/events/"
]

async def crawl_all_sites():
    """Краулим все сайты из списка"""
    pipeline = EventCrawlerPipeline()
    results = []
    
    for url in TARGET_SITES:
        try:
            result = await pipeline.process_url(url)
            results.append({"url": url, **result})
            # Пауза между запросами
            await asyncio.sleep(5)
        except Exception as e:
            print(f"Ошибка при обработке {url}: {e}")
            results.append({"url": url, "error": str(e)})
    
    return results

def run_scheduled_crawl():
    """Запуск по расписанию"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Запуск scheduled краулинга...")
    asyncio.run(crawl_all_sites())
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Краулинг завершен")

if __name__ == "__main__":
    # Настраиваем расписание
    schedule.every().day.at("08:00").do(run_scheduled_crawl)
    schedule.every().day.at("14:00").do(run_scheduled_crawl)
    schedule.every().day.at("20:00").do(run_scheduled_crawl)
    
    print("Мультимодальный краулер событий запущен")
    print("Расписание: 8:00, 14:00, 20:00")
    print("Для ручного запуска нажмите Ctrl+C и запустите `python main.py --now")
    
    # Бесконечный цикл для выполнения по расписанию
    while True:
        schedule.run_pending()
        time.sleep(60)

Критические нюансы и возможные ошибки

1. Проблема с памятью — GLM-4.6V может «съедать» всю видеопамять. Решение: используйте квантованные версии (4bit, 5bit) и ограничивайте batch size.
2. Галлюцинации модели — LLM иногда выдумывает события. Добавляйте валидацию: проверяйте формат дат, наличие реальных мест.
3. Блокировка краулера — некоторые сайты блокируют Playwright. Используйте ротацию user-agent, добавляйте задержки.
4. Скорость обработки — анализ изображений медленный. Параллелизируйте обработку с помощью asyncio или multiprocessing.
5. Качество скриншотов — на динамических сайтах элементы могут не успеть загрузиться. Увеличьте timeout в wait_until.

Оптимизация и масштабирование

Когда система работает, можно задуматься об оптимизации:

• Кэширование эмбеддингов — не пересчитывайте эмбеддинги для одинаковых текстов
• Пакетная обработка — обрабатывайте несколько изображений за один вызов модели
• Использование GPU кластера — распределяйте нагрузку между несколькими GPU. Читайте стратегии масштабирования локальных LLM
• Оптимизация промптов — экспериментируйте с промптами для повышения точности

Часто задаваемые вопросы (FAQ)

❓ Можно ли использовать другие мультимодальные модели?

Да, кроме GLM-4.6V можно использовать LLaVA, Qwen-VL, или даже собрать пайплайн из отдельных моделей: одна для текста (например, Llama 3.3), другая для изображений (BLIP). Но GLM-4.6V хорош именно интеграцией в одну модель.

❓ Что делать, если не хватает видеопамяти?

Используйте квантованные версии моделей (4bit, 5bit), уменьшайте размер контекста, или запускайте модель на CPU (медленнее, но работает). Также можно использовать фреймворки с оптимизацией памяти.

❓ Как добавить поддержку новых типов событий?

Достаточно расширить промпты в классе MultimodalAnalyzer. Добавьте новые категории в prompt и обучите модель на примерах (few-shot learning).

❓ Как избежать блокировки сайтами?

Используйте ротацию IP (если есть доступ к прокси), случайные задержки между запросами, меняйте user-agent. Для сложных случаев рассмотрите использование headless браузера с реальным Chrome профилем.

❓ Можно ли коммерциализировать такую систему?

Да, но будьте осторожны с авторскими правами и условиями использования сайтов. Можно предлагать сервис мониторинга событий для бизнеса, или использовать систему для внутренних нужд компании. Для микро-платежей за использование AI-агентов изучите модели микроплатежей для AI-агентов.

Заключение

Мы построили полностью автономную систему мультимодального краулинга событий, которая работает локально, не зависит от облачных API и полностью контролируется вами. Это лишь отправная точка — систему можно расширять:

• Добавить анализ PDF-флаеров и брошюр
• Интегрировать с календарями (Google Calendar, Outlook)
• Создать Telegram-бота для уведомлений о событиях
• Добавить рекомендательную систему на основе векторных эмбеддингов

Локальные LLM открывают новые возможности для создания приватных, контролируемых и экономичных AI-систем. Начните с этого проекта, экспериментируйте с разными моделями и архитектурами, и вы найдете множество применений для мультимодального AI в своих задачах.