Локальная LLM + Vertex AI Search: гибридная архитектура для заземлённого поиска без парсинга

Почему никто не говорит правду про поиск в 2026 году?

Есть грязный секрет, о котором молчат все провайдеры поисковых API. Ты платишь за "заземлённый" поиск, а получаешь обычный парсинг веб-страниц с накруткой релевантности. Google называет это Vertex AI Search, но под капотом - всё те же старые методы, которые требуют тонны предобработки и выжимают из тебя деньги за каждый запрос.

А что если я скажу, что можно получить лучшее из двух миров? Локальная приватность твоей LLM плюс промышленная мощь поиска Google. И всё это без парсинга, без предварительной обработки документов, без необходимости загружать терабайты данных в облако.

Зачем это вообще нужно? (Спойлер: не для всех)

Представь: у тебя есть Qwen3 VL 4b, работающая локально на твоём RTX 4080 с 16 ГБ памяти. Она умная, приватная, но знает только то, что ты её научил. А тебе нужно отвечать на вопросы про текущие события, актуальные цены, последние обновления ПО.

Классический RAG? Забудь. Ты будешь парсить сайты, чистить HTML, векторизовать, обновлять индексы. Месяц работы на подготовку данных. А через неделю всё устареет.

Вот где появляется гибридная архитектура: локальная LLM генерирует финальный ответ, а Vertex AI Search подкидывает ей свежие, релевантные факты. Как напарник, который всегда в курсе последних новостей.

Что тебе понадобится (и сколько это стоит)

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

1 Подготавливаем локальную LLM: Qwen3 VL 4b не как все

Первая ошибка, которую совершают 90% разработчиков: пытаются запустить Qwen3 VL 4b "из коробки". Не работает. Модель требует специфичной настройки контекста для работы с внешними данными.

Вот как НЕ надо делать:

# ПЛОХОЙ ПРИМЕР - не копируй это
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen3-VL-4b")
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen3-VL-4b")

# И дальше пытаться засунуть в неё результаты поиска...

Почему плохо? Потому что Qwen3 VL 4b на 28.01.2026 имеет контекстное окно в 32К токенов, но без правильного промптинга она будет игнорировать внешние данные. Модель продолжит генерировать ответы из своих тренировочных данных.

Вот правильный подход:

# ХОРОШИЙ ПРИМЕР
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TextStreamer

# Загружаем последнюю версию на 28.01.2026
model_name = "Qwen/Qwen3-VL-4b-Instruct"  # Инструктивная версия!

tokenizer = AutoTokenizer.from_pretrained(
    model_name,
    trust_remote_code=True,
    padding_side="left"  # Критично для инференса
)

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    device_map="auto",
    trust_remote_code=True
)

# Системный промпт для работы с внешними данными
system_prompt = """Ты - ассистент, который отвечает на вопросы, используя предоставленные факты из поиска.
Если в предоставленных фактах нет ответа - скажи об этом честно.
Не выдумывай информацию. Используй только то, что предоставлено ниже.

Факты из поиска:
{search_results}

Вопрос пользователя: {user_question}

Ответ (основанный только на фактах выше):"""

2 Настраиваем Vertex AI Search: не попадись на эти грабли

Google Cloud Console - место, где даже опытные разработчики теряют по три часа. Интерфейс меняется каждые полгода, кнопки перемещаются, а нужные настройки прячутся в неочевидных местах.

Шаг 1: Создаём проект (если ещё нет)

# Устанавливаем gcloud CLI
curl https://sdk.cloud.google.com | bash
gcloud init

# Создаём новый проект
PROJECT_ID="your-project-$(date +%s)"
gcloud projects create $PROJECT_ID

# Устанавливаем его как текущий
gcloud config set project $PROJECT_ID

Шаг 2: Включаем API (здесь первая ловушка)

# НЕ ТОЛЬКО Vertex AI API!
gcloud services enable \
    aiplatform.googleapis.com \
    discoveryengine.googleapis.com \
    --project=$PROJECT_ID

Да, нужны два API: aiplatform (общий) и discoveryengine (специфичный для поиска). Если включишь только первый - будешь получать 404 ошибки при вызовах поиска.

Шаг 3: Создаём сервисный аккаунт и ключ

# Создаём сервисный аккаунт
SA_NAME="vertex-search-sa"
gcloud iam service-accounts create $SA_NAME \
    --display-name="Vertex AI Search Service Account"

# Даём права
SA_EMAIL="$SA_NAME@$PROJECT_ID.iam.gserviceaccount.com"
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/aiplatform.user"

gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$SA_EMAIL" \
    --role="roles/discoveryengine.viewer"

# Создаём ключ
gcloud iam service-accounts keys create vertex-key.json \
    --iam-account=$SA_EMAIL

3 Пишем интеграцию: где спотыкаются даже senior'ы

А теперь самое интересное - код, который соединяет локальную модель с облачным поиском. Основная проблема: асинхронность. Поиск в Google может занимать от 200 мс до 2 секунд, а локальная модель тоже не мгновенная.

Неправильный подход (блокирующий):

# ПЛОХО - блокирует всё
search_results = vertex_search(query)  # Ждём 2 секунды
answer = local_llm.generate(search_results, query)  # Ждём ещё 5 секунд
return answer  # Пользователь уже ушёл

Правильный подход (асинхронный с таймаутами):

import asyncio
import aiohttp
from google.auth.transport.requests import Request
from google.oauth2 import service_account
from typing import List, Dict, Optional
import json

class HybridSearchAssistant:
    def __init__(self, credentials_path: str, project_id: str, location: str = "global"):
        self.credentials = service_account.Credentials.from_service_account_file(
            credentials_path,
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
        self.project_id = project_id
        self.location = location
        self.search_url = f"https://discoveryengine.googleapis.com/v1beta/projects/{project_id}/locations/{location}/collections/default_collection/dataStores/default_data_store/servingConfigs/default_search:search"
    
    async def search_vertex(self, query: str, max_results: int = 5) -> List[Dict]:
        """Асинхронный поиск через Vertex AI Search"""
        headers = {
            "Authorization": f"Bearer {self.credentials.token}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "query": query,
            "pageSize": max_results,
            "queryExpansionSpec": {
                "condition": "AUTO"  # Авторасширение запросов
            },
            "spellCorrectionSpec": {
                "mode": "AUTO"  # Автоисправление опечаток
            },
            "contentSearchSpec": {
                "snippetSpec": {
                    "maxSnippetCount": 3,
                    "returnSnippet": True
                },
                "summarySpec": {
                    "summaryResultCount": 3,
                    "includeCitations": True,
                    "ignoreAdversarialQuery": True
                }
            }
        }
        
        async with aiohttp.ClientSession() as session:
            try:
                # Таймаут: 3 секунды на поиск
                async with session.post(
                    self.search_url,
                    headers=headers,
                    json=payload,
                    timeout=aiohttp.ClientTimeout(total=3.0)
                ) as response:
                    if response.status == 200:
                        data = await response.json()
                        return self._extract_search_results(data)
                    else:
                        error_text = await response.text()
                        print(f"Vertex Search error: {error_text}")
                        return []
            except asyncio.TimeoutError:
                print("Vertex Search timeout")
                return []
    
    def _extract_search_results(self, data: Dict) -> List[Dict]:
        """Извлекаем и структурируем результаты поиска"""
        results = []
        
        if "results" in data:
            for item in data["results"]:
                result = {
                    "title": item.get("document", {}).get("derivedStructData", {}).get("title", ""),
                    "snippet": item.get("document", {}).get("derivedStructData", {}).get("snippets", [{}])[0].get("snippet", ""),
                    "link": item.get("document", {}).get("derivedStructData", {}).get("link", ""),
                    "score": item.get("relevanceScore", 0.0)
                }
                # Фильтруем пустые сниппеты
                if result["snippet"]:
                    results.append(result)
        
        # Сортируем по релевантности
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:5]  # Возвращаем топ-5
    
    async def generate_answer(self, query: str) -> str:
        """Главный метод: поиск + генерация"""
        # Параллельно обновляем токен и делаем поиск
        self.credentials.refresh(Request())
        search_task = asyncio.create_task(self.search_vertex(query))
        
        # Ждём результаты поиска (макс 3 секунды)
        search_results = await search_task
        
        if not search_results:
            return "Извините, не удалось найти информацию по вашему запросу."
        
        # Форматируем результаты для промпта
        formatted_results = "\n".join([
            f"{i+1}. {r['title']}: {r['snippet']}"
            for i, r in enumerate(search_results)
        ])
        
        # Готовим промпт для локальной LLM
        prompt = system_prompt.format(
            search_results=formatted_results,
            user_question=query
        )
        
        # Генерация ответа (локально)
        inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=8192)
        inputs = inputs.to(model.device)
        
        with torch.no_grad():
            outputs = model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=0.3,  # Низкая температура для фактологичности
                do_sample=True,
                top_p=0.9,
                repetition_penalty=1.1,
                pad_token_id=tokenizer.pad_token_id
            )
        
        answer = tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # Вырезаем только ответ (после последнего двоеточия в промпте)
        answer_start = answer.find("Ответ (основанный только на фактах выше):") + len("Ответ (основанный только на фактах выше):")
        clean_answer = answer[answer_start:].strip()
        
        return clean_answer

Типичные ошибки и как их избежать

После десятка таких интеграций я собрал коллекцию грабель, на которые наступают все:

• Ошибка 1: Токены Google Cloud устаревают через час. Решение: автоматическое обновление перед каждым запросом или использование долгоживущих токенов.
• Ошибка 2: Vertex AI Search возвращает HTML в сниппетах. Решение: включить "summarySpec" в запросе - Google сам сгенерирует чистый текст.
• Ошибка 3: Локальная модель "забывает" промпт и генерирует общие ответы. Решение: использовать инструктивные версии моделей и жёсткие разделители в промпте.
• Ошибка 4: Запросы к Google блокируются фаерволом. Решение: использовать официальные SDK вместо raw HTTP запросов.

Производительность: чего ждать от гибридной системы

Давай без прикрас. На RTX 4080 (16 ГБ) с Qwen3 VL 4b:

• Холодный старт: 8-12 секунд (загрузка модели в память)
• Поиск в Vertex AI: 200-800 мс (зависит от сложности запроса)
• Генерация ответа: 2-4 секунды (для 512 токенов)
• Общее время ответа: 3-6 секунд

Это медленнее, чем чистый Vertex AI (500 мс), но быстрее, чем RAG с собственным парсингом (где нужно сначала найти, потом распарсить, потом векторизовать).

Память: Qwen3 VL 4b занимает ~8 ГБ в FP16, остальное - для контекста и кэша. На 16 ГБ GPU можно комфортно работать с контекстом до 16К токенов.

Когда эта архитектура не подойдёт

Я должен быть честен: гибридный подход - не серебряная пуля. Он не сработает, если:

1. Тебе нужна обработка в реальном времени (<100 мс)
2. У тебя нет стабильного интернета (поиск требует подключения)
3. Ты работаешь с highly sensitive данными, которые нельзя отправлять даже в Google
4. Бюджет ограничен $10 в месяц (Vertex AI Search сожрёт их за 5000 запросов)

В таких случаях смотри в сторону полностью локальных решений. У меня есть статья про Meeting-LLM, где всё работает офлайн.

Что будет дальше? (Мои прогнозы на 2026-2027)

К концу 2026 года Google, скорее всего, представит нативный SDK для гибридных архитектур. Уже сейчас в бета-тесте есть "Vertex AI Hybrid Endpoints", которые позволяют запускать кастомные модели рядом с поиском.

Локальные модели продолжат уменьшаться в размерах при сохранении качества. Ожидаю появления 3-4 миллиардных моделей, которые по качеству будут как сегодняшние 7-миллиардные.

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

Гибридная архитектура - это не про идеальное решение. Это про практический компромисс между стоимостью, приватностью и актуальностью. Она позволяет строить системы, которые вчера были доступны только гигантам вроде Google или Microsoft.

А теперь иди и собери свою первую гибридную систему. Первые 500 запросов в Vertex AI Search бесплатные - хватит, чтобы понять, подходит ли тебе этот подход.