RAG для тех, кто разочаровался: почему retrieval ломается и как это починить

from langchain.text_splitter import RecursiveCharacterTextSplitter
# Стандартный подход — работает, но грубо
basic_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
# Лучше: разделители по приоритету
# Сначала пробуем разрезать по двойному переносу строки (между параграфами),
# потом по одинарному, потом по предложению, потом по символам
smart_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)

import re
def split_by_sections(text: str, max_chunk_size: int = 2000) -> list[dict]:
    """Нарезка по заголовкам разделов (markdown-стиль)."""
    sections = re.split(r'\n(#{1,3}\s+.+)\n', text)
    chunks = []
    current_header = ""
    for i, section in enumerate(sections):
        if re.match(r'^#{1,3}\s+', section):
            current_header = section.strip()
            continue
        content = section.strip()
        if not content:
            continue
        # Если секция слишком большая — разбиваем дальше
        if len(content) > max_chunk_size:
            sub_chunks = smart_splitter.split_text(content)
            for j, sub in enumerate(sub_chunks):
                chunks.append({
                    "text": f"{current_header}\n\n{sub}",
                    "metadata": {
                        "section": current_header,
                        "part": j + 1,
                    }
                })
        else:
            chunks.append({
                "text": f"{current_header}\n\n{content}",
                "metadata": {"section": current_header}
            })
    return chunks

from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
    def __init__(self, chunks: list[dict], embeddings_model):
        self.chunks = chunks
        self.texts = [c["text"] for c in chunks]
        # BM25 для keyword search
        tokenized = [text.lower().split() for text in self.texts]
        self.bm25 = BM25Okapi(tokenized)
        # Эмбеддинги для semantic search
        self.embeddings = embeddings_model.encode(self.texts)
        self.model = embeddings_model
    def search(self, query: str, top_k: int = 5, alpha: float = 0.5) -> list[dict]:
        # BM25 scores
        bm25_scores = self.bm25.get_scores(query.lower().split())
        bm25_norm = bm25_scores / (bm25_scores.max() + 1e-6)
        # Semantic scores
        query_emb = self.model.encode(query)
        cos_scores = np.dot(self.embeddings, query_emb) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_emb) + 1e-6
        )
        cos_norm = (cos_scores - cos_scores.min()) / (cos_scores.max() - cos_scores.min() + 1e-6)
        # Комбинируем: alpha * semantic + (1-alpha) * keyword
        combined = alpha * cos_norm + (1 - alpha) * bm25_norm
        top_indices = np.argsort(combined)[::-1][:top_k]
        return [self.chunks for i in top_indices]

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def search_with_rerank(query: str, retriever: HybridRetriever, top_k: int = 5):
    # Грубый поиск: 20 кандидатов
    candidates = retriever.search(query, top_k=20)
    # Reranking: cross-encoder оценивает пару (query, candidate)
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    # Сортируем по score от reranker
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [c for c, s in ranked[:top_k]]

eval_dataset = [
    {
        "question": "Какая неустойка за просрочку поставки?",
        "expected_chunk_id": "contract_7_3",  # ID чанка с пунктом 7.3
    },
    {
        "question": "Какой срок оплаты по договору?",
        "expected_chunk_id": "contract_4_1",
    },
    # ... ещё 48–98 вопросов
]
def evaluate_retrieval(retriever, eval_dataset, top_k=5):
    hits = 0
    for item in eval_dataset:
        results = retriever.search(item["question"], top_k=top_k)
        result_ids = [r["metadata"].get("chunk_id") for r in results]
        if item["expected_chunk_id"] in result_ids:
            hits += 1
    recall = hits / len(eval_dataset)
    return recall
# Базовый retrieval
recall_basic = evaluate_retrieval(basic_retriever, eval_dataset)
print(f"Basic retrieval recall@5: {recall_basic:.2%}")
# После улучшений (гибридный + reranking)
recall_improved = evaluate_retrieval(improved_retriever, eval_dataset)
print(f"Improved retrieval recall@5: {recall_improved:.2%}")

• 27 мая, 20:00. «Мифы про ИИ-агентов: что реально работает в 2026 году». Записаться
  
• 15 июня, 20:00. «Интеграция ИИ-агентов в рабочую разработку: обвязка агента навыками и MCP». Записаться