Разработка RAG с векторной базой данных Pinecone

Разработка RAG с векторной базой данных Pinecone

Pinecone — managed векторная база данных с REST/gRPC API, автоматическим масштабированием и поддержкой гибридного поиска (sparse + dense). Не требует управления инфраструктурой, легко масштабируется от прототипа до миллионов векторов. Pinecone Serverless (с 2024 года) позволяет работать без предварительного резервирования ресурсов — платите только за фактические операции.

Инициализация и создание индекса

from pinecone import Pinecone, ServerlessSpec
from openai import OpenAI

pc = Pinecone(api_key="...")

# Создание serverless индекса
pc.create_index(
    name="corporate-knowledge-base",
    dimension=3072,        # text-embedding-3-large
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

index = pc.Index("corporate-knowledge-base")

Индексация документов с метаданными

from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
import hashlib

embeddings_model = OpenAIEmbeddings(model="text-embedding-3-large")

def index_documents(documents: list, batch_size: int = 100):
    splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
    chunks = splitter.split_documents(documents)

    # Батчевая индексация
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]

        texts = [c.page_content for c in batch]
        vectors = embeddings_model.embed_documents(texts)

        # Формируем записи для Pinecone
        records = []
        for chunk, vector in zip(batch, vectors):
            doc_id = hashlib.md5(chunk.page_content.encode()).hexdigest()
            records.append({
                "id": doc_id,
                "values": vector,
                "metadata": {
                    "text": chunk.page_content,
                    "source": chunk.metadata.get("source", ""),
                    "page": chunk.metadata.get("page", 0),
                    "doc_type": chunk.metadata.get("doc_type", "general"),
                    "date": chunk.metadata.get("date", ""),
                }
            })

        index.upsert(vectors=records)
        print(f"Indexed batch {i//batch_size + 1}: {len(records)} chunks")

Запрос с фильтрацией по метаданным

def rag_query(
    query: str,
    doc_type_filter: str = None,
    top_k: int = 5
) -> dict:

    # Embedding запроса
    query_vector = embeddings_model.embed_query(query)

    # Формируем фильтр
    filter_dict = {}
    if doc_type_filter:
        filter_dict["doc_type"] = {"$eq": doc_type_filter}

    # Поиск
    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True,
        filter=filter_dict if filter_dict else None
    )

    # Формируем контекст
    context_chunks = []
    for match in results["matches"]:
        context_chunks.append({
            "text": match["metadata"]["text"],
            "source": match["metadata"]["source"],
            "score": match["score"]
        })

    return context_chunks

Hybrid Search в Pinecone

Pinecone поддерживает гибридный поиск (dense + sparse) через встроенный BM25:

from pinecone_text.sparse import BM25Encoder

# Обучение BM25 на корпусе документов
bm25 = BM25Encoder()
bm25.fit(all_texts)

def hybrid_query(query: str, alpha: float = 0.5, top_k: int = 5) -> list:
    """
    alpha=1.0: только dense
    alpha=0.0: только sparse (BM25)
    alpha=0.5: равный вес обоих
    """
    # Dense вектор
    dense_vector = embeddings_model.embed_query(query)

    # Sparse вектор (BM25)
    sparse_vector = bm25.encode_queries(query)

    results = index.query(
        vector=dense_vector,
        sparse_vector=sparse_vector,
        top_k=top_k,
        include_metadata=True,
        alpha=alpha,
    )
    return results["matches"]

Практический кейс: корпоративная база знаний ретейлера

Масштаб: 45 000 SKU с описаниями, 3200 страниц регламентов, 800 FAQ-записей. Суммарно ~180 000 векторов.

Конфигурация: Pinecone Serverless (aws/us-east-1), dimension=1536 (text-embedding-3-small для экономии), metric=cosine.

Паттерн использования: 15 000 запросов/день, пиковая нагрузка 200 RPS в часы распродаж.

Результаты:

• Latency P95 для retrieval: 180мс
• Latency P95 для полного RAG ответа: 2.1с (включая GPT-4o-mini)
• Стоимость Pinecone: ~$80/месяц (Serverless)
• Context recall (нашли нужный документ): 0.87
• Answer accuracy (LLM-judge): 0.83

Оптимизации:

• Namespace separation: товары/регламенты/FAQ в отдельных namespace — позволяет фильтровать без overhead
• Metadata-only queries: для ряда запросов достаточно фильтра по метаданным без векторного поиска
• Cache популярных запросов: Redis кэш для топ-500 частых вопросов (~30% hit rate)

Сроки

• Настройка Pinecone + ingestion pipeline: 3–5 дней
• RAG-пайплайн с оценкой качества: 1–2 недели
• Оптимизация и production: 1–2 недели
• Итого: 2–5 недель