AI-система Search-as-a-Service (поисковый AI-сервис для клиентов)

AI Search as a Service: поиск как платформа

Строить поиск с нуля для каждого продукта — дорого и долго. Search as a Service — это готовая поисковая платформа с AI-слоем, которую команды подключают через API или SDK, не разбираясь в векторных базах данных и re-ranking моделях.

Типичный сценарий: у компании 5 продуктов, в каждом нужен поиск по каталогу / документам / контенту. Без платформы — 5 независимых реализаций, 5 раз настраивать индексы, 5 раз платить за GPU для embedding-модели. С платформой — один shared сервис, разные индексы (tenants), единый API.

Архитектура multi-tenant поисковой платформы

from fastapi import FastAPI, Header, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
import uuid

app = FastAPI(title="Search as a Service")

class IndexConfig(BaseModel):
    name: str
    embedding_model: str = "intfloat/multilingual-e5-large"
    chunk_size: int = 512
    chunk_overlap: int = 64
    language: str = "ru"
    reranker_enabled: bool = True

class SearchRequest(BaseModel):
    query: str
    index_name: str
    filters: Optional[dict] = None
    top_k: int = 10
    mode: str = "hybrid"  # "vector" | "keyword" | "hybrid"
    generate_answer: bool = False

class SearchService:
    def __init__(self):
        self.tenant_indexes = {}  # tenant_id → {index_name → index}
        self.embedding_models = {}  # model_name → loaded_model
        self.reranker = self._load_reranker()

    async def create_index(self, tenant_id: str, config: IndexConfig):
        """Создаёт изолированный индекс для тенанта"""
        collection_name = f"{tenant_id}_{config.name}"
        # Каждый тенант — отдельная коллекция в Qdrant
        # с собственными payload-фильтрами
        self.qdrant.create_collection(
            collection_name=collection_name,
            vectors_config=VectorParams(
                size=self._get_vector_size(config.embedding_model),
                distance=Distance.COSINE
            )
        )
        return {"index_id": collection_name, "status": "created"}

    async def search(
        self,
        tenant_id: str,
        request: SearchRequest
    ) -> dict:
        collection = f"{tenant_id}_{request.index_name}"

        if request.mode == "hybrid":
            results = await self._hybrid_search(
                collection, request.query,
                request.filters, request.top_k
            )
        elif request.mode == "vector":
            results = await self._vector_search(
                collection, request.query, request.top_k
            )
        else:
            results = await self._keyword_search(
                collection, request.query, request.top_k
            )

        if request.generate_answer and results:
            answer = await self._generate_answer(request.query, results)
            return {"results": results, "answer": answer}

        return {"results": results}

SDK для команд-потребителей

# pip install search-platform-sdk
from search_platform import SearchClient

client = SearchClient(
    api_key="sk-...",
    base_url="https://search.internal.company.com"
)

# Индексация документов
client.index.upload(
    index_name="product-catalog",
    documents=[
        {"id": "p001", "title": "Ноутбук Dell XPS", "description": "...",
         "price": 89999, "category": "laptops"},
        # ...
    ]
)

# Поиск
results = client.search(
    index_name="product-catalog",
    query="тонкий ноутбук для работы с видео",
    filters={"price": {"lte": 100000}, "category": "laptops"},
    top_k=5,
    generate_answer=True
)

print(results.answer)   # "На основе вашего запроса рекомендую..."
print(results.items)    # список документов с релевантностью

Rate limiting и мониторинг

from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import redis.asyncio as redis

# Лимиты по тенантам
TENANT_LIMITS = {
    "free": "100/minute",
    "pro": "1000/minute",
    "enterprise": "unlimited"
}

@app.post("/search")
@limiter.limit(get_tenant_limit)  # динамический лимит по плану
async def search_endpoint(
    request: SearchRequest,
    x_api_key: str = Header(...),
    tenant = Depends(authenticate_tenant)
):
    return await search_service.search(tenant.id, request)

Billing и usage tracking

Каждый поиск и каждый embedding-запрос логируются в TimescaleDB:

CREATE TABLE search_usage (
    id          BIGSERIAL PRIMARY KEY,
    tenant_id   TEXT NOT NULL,
    index_name  TEXT NOT NULL,
    query_hash  TEXT,           -- хэш для анонимизации
    mode        TEXT,
    latency_ms  INTEGER,
    result_count INTEGER,
    answer_generated BOOLEAN,
    tokens_used  INTEGER,       -- для LLM-ответа
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- Гипертаблица TimescaleDB для эффективного time-range запросов
SELECT create_hypertable('search_usage', 'created_at');

Кейс: SaaS-компания с 12 продуктами мигрировала со своих разрозненных Elasticsearch-инстансов на единую платформу. 3 команды подключились через SDK за 1 день (без понимания векторных баз и embedding-моделей). Стоимость инфраструктуры снизилась на 35% за счёт shared GPU для embedding. Среднее время ответа поиска: 280 мс P50, 650 мс P99 на корпусе 2M документов.

SLA и параметры платформы

Сроки: базовая платформа (API + индексация + гибридный поиск): 6–8 недель; с генерацией ответов, SDK и биллингом: 3–4 месяца.