Проектування архітектури RAG-пайплайну

Проектування архітектури RAG-пайплайну

Архітектура RAG-пайплайну визначає якість, масштабованість та вартість всієї системи. Базова RAG «працює» за день, але production-ready система з надійним retrieval, моніторингом та керованою вартістю вимагає ретельного проектування.

Компоненти сучасного RAG-пайплайну

┌─────────────────────────────────────────────────────┐
│                 INGESTION PIPELINE                   │
│  Sources → Loaders → Parsers → Chunkers → Embedder  │
│           → Metadata Extractor → Vector Store        │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│                 RETRIEVAL PIPELINE                   │
│  Query → Query Transformer → Multi-Index Search     │
│        → Reranker → Context Assembler               │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│                GENERATION PIPELINE                   │
│  Context + Query → Prompt Builder → LLM             │
│                  → Response Validator → User         │
└─────────────────────────────────────────────────────┘

Ingestion Pipeline: архітектурні рішення

Document Loaders: вибір завантажувачів критичний для якості. PDF з таблицями вимагає pdfplumber або LlamaParse, а не PyPDF2. Word документи — python-docx, HTML — BeautifulSoup з користувацькими правилами очистки.

from llama_parse import LlamaParse
from langchain_community.document_loaders import (
    PyPDFLoader, UnstructuredWordDocumentLoader,
    ConfluenceLoader, NotionDBLoader
)

# Для складних PDF (таблиці, колони, зображення)
parser = LlamaParse(
    api_key="...",
    result_type="markdown",  # Зберігає структуру таблиць
    language="uk",
)

# Конфігуровий пайплайн завантаження
LOADERS = {
    ".pdf": lambda path: LlamaParse().load_data(path),
    ".docx": lambda path: UnstructuredWordDocumentLoader(path).load(),
    ".html": lambda path: custom_html_loader(path),
}

Metadata enrichment: збагачення чанків метаданими критично для фільтрації та атрибуції:

def enrich_chunk_metadata(chunk, source_doc):
    """Додає структуровані метаданні до чанку"""
    chunk.metadata.update({
        "source": source_doc.metadata.get("source"),
        "page": source_doc.metadata.get("page"),
        "doc_type": detect_doc_type(source_doc),     # "contract", "regulation", "faq"
        "department": extract_department(source_doc),
        "date": extract_date(source_doc),
        "version": extract_version(source_doc),
        "chunk_index": chunk.metadata.get("chunk_index"),
        "parent_chunk_id": chunk.metadata.get("parent_id"),
    })
    return chunk

Retrieval Pipeline: стратегії

Sparse + Dense Hybrid Search:

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector, NamedVector

# Гібридний пошук у Qdrant: BM25 sparse + embedding dense
def hybrid_search(query: str, top_k: int = 10) -> list:
    # Dense embedding
    dense_vector = embedder.embed_query(query)

    # Sparse (BM25) через SPLADE або FastEmbed
    sparse_vector = sparse_encoder.encode(query)

    results = client.query_points(
        collection_name="docs",
        prefetch=[
            {"query": dense_vector, "using": "dense", "limit": 30},
            {"query": SparseVector(indices=sparse_vector.indices,
                                   values=sparse_vector.values),
             "using": "sparse", "limit": 30},
        ],
        query=fusion,  # RRF (Reciprocal Rank Fusion)
        limit=top_k,
    )
    return results

Reranking Pipeline:

from flashrank import Ranker, RerankRequest

ranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2")

def rerank_results(query: str, candidates: list[str]) -> list[str]:
    rerank_request = RerankRequest(
        query=query,
        passages=[{"id": i, "text": c} for i, c in enumerate(candidates)]
    )
    results = ranker.rerank(rerank_request)
    # Сортуємо за score, беремо top-5
    top_passages = [candidates[r["id"]] for r in sorted(results, key=lambda x: -x["score"])[:5]]
    return top_passages

Query Transformation: поліпшення запиту перед пошуком

Погано сформульований запит = поганий retrieval. Трансформації запиту:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Multi-Query: генеруємо 3 перефразування запиту
def multi_query_transform(original_query: str) -> list[str]:
    response = llm.invoke(f"""Згенеруй 3 різних перефразування наступного питання.
Кожен варіант має шукати ту ж інформацію, але іншими словами.
Повернути JSON-список рядків.

Питання: {original_query}""")
    queries = json.loads(response.content)
    return [original_query] + queries  # Оригінал + 3 перефразування

# Step-back prompting: абстрагуємося до більш загального питання
def step_back_transform(specific_query: str) -> str:
    response = llm.invoke(f"""Сформулюй більш загальне питання, відповідь на яке
допомогла б відповісти на конкретне питання: "{specific_query}"
Поверни лише питання, без пояснень.""")
    return response.content

Архітектура з кількома індексами

Для систем з різнорідними джерелами даних ефективніше використовувати окремі індекси:

class MultiIndexRAG:
    def __init__(self):
        self.indexes = {
            "contracts": QdrantRetriever(collection="contracts"),
            "regulations": QdrantRetriever(collection="regulations"),
            "faq": QdrantRetriever(collection="faq"),
            "procedures": QdrantRetriever(collection="procedures"),
        }
        self.router = QueryRouter()  # Класифікатор запитів

    def retrieve(self, query: str, top_k: int = 5) -> list:
        # Визначаємо релевантні індекси
        relevant_indexes = self.router.route(query)

        # Паралельний пошук по всіх релевантних індексах
        all_results = []
        for index_name in relevant_indexes:
            results = self.indexes[index_name].retrieve(query, k=top_k)
            for r in results:
                r.metadata["source_index"] = index_name
            all_results.extend(results)

        # Reranking об'єднаних результатів
        return rerank_results(query, all_results)[:top_k]

Моніторинг якості retrieval

# Трасування кожного запиту для аналізу
import opentelemetry as otel

def traced_retrieval(query: str, span_name: str = "rag_retrieval"):
    with otel.trace.get_tracer(__name__).start_as_current_span(span_name) as span:
        start_time = time.time()
        results = retriever.retrieve(query)
        latency = time.time() - start_time

        span.set_attributes({
            "query.length": len(query),
            "results.count": len(results),
            "results.top_score": results[0].score if results else 0,
            "retrieval.latency_ms": latency * 1000,
        })
        return results

Строки проектування та розробки

• Проектування архітектури: 1 тиждень
• Базовий ingestion pipeline: 1–2 тижні
• Advanced retrieval (hybrid search, reranking): 2–3 тижні
• Evaluation framework: 1–2 тижні
• Production hardening: 1–2 тижні
• Всього: 6–10 тижнів