Reranking: cross-encoder модели

pip install sentence-transformers

from sentence_transformers import CrossEncoder

# Загружаем модель (скачивается при первом вызове)
# Для English: "cross-encoder/ms-marco-MiniLM-L-6-v2"
# Для мультиязычного: "BAAI/bge-reranker-v2-m3"
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

query = "как обработать ошибку подключения к Redis"

# Кандидаты от первичного ANN-поиска
candidates = [
    "Redis: архитектура, типы данных, персистентность",
    "Работа с Redis в Python: подключение, основные команды",
    "ConnectionError при подключении к Redis: диагностика",
    "Redis Sentinel: настройка высокой доступности",
    "Retry-логика и backoff при ошибках сети в redis-py",
    "Redis Cluster: шардирование и репликация",
    "Мониторинг Redis с помощью Prometheus и Grafana",
]

# Формируем пары (query, doc) для каждого кандидата
pairs = [[query, doc] for doc in candidates]

# Predict возвращает массив скоров для каждой пары
scores = reranker.predict(pairs)

# Сортируем кандидатов по убыванию скора
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

print(f"Query: {query}\n")
for rank, (doc, score) in enumerate(ranked, 1):
    print(f"#{rank:2d}  [{score:6.3f}]  {doc}")

Query: как обработать ошибку подключения к Redis

#1  [ 9.841]  Retry-логика и backoff при ошибках сети в redis-py
#2  [ 8.312]  ConnectionError при подключении к Redis: диагностика
#3  [ 3.124]  Работа с Redis в Python: подключение, основные команды
#4  [ 0.871]  Redis: архитектура, типы данных, персистентность
#5  [-1.203]  Redis Sentinel: настройка высокой доступности
#6  [-2.891]  Мониторинг Redis с помощью Prometheus и Grafana
#7  [-4.512]  Redis Cluster: шардирование и репликация

import torch
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, candidates: list[str], top_k: int = 5, min_score: float = 0.1):
    """
    Reranker с нормализацией sigmoid и пороговым отсечением.

    Args:
        query:      Запрос пользователя
        candidates: Список текстов-кандидатов
        top_k:      Максимальное количество результатов
        min_score:  Минимальный порог релевантности (0–1)

    Returns:
        Список (doc, score) отсортированный по убыванию
    """
    if not candidates:
        return []

    pairs = [[query, doc] for doc in candidates]

    # activation_fct=torch.sigmoid нормализует в [0, 1]
    scores = reranker.predict(pairs, activation_fct=torch.sigmoid)

    ranked = sorted(
        zip(candidates, scores.tolist()),
        key=lambda x: x[1],
        reverse=True,
    )

    # Применяем порог и ограничение top_k
    return [(doc, score) for doc, score in ranked if score >= min_score][:top_k]


# Пример
results = rerank(
    "как обработать ошибку подключения к Redis",
    candidates,
    top_k=3,
    min_score=0.3,  # отсекаем нерелевантные (score < 0.3)
)

for doc, score in results:
    print(f"[{score:.3f}] {doc}")

pip install cohere

import cohere
import os

co = cohere.Client(os.environ["COHERE_API_KEY"])

query = "Python кеширование через декораторы"

candidates = [
    "Декораторы в Python: полное руководство по синтаксису",
    "functools.lru_cache: кеширование с LRU-стратегией",
    "Python метапрограммирование: декораторы классов",
    "Кеширование HTTP-ответов в Django",
    "Как написать собственный декоратор кеша с TTL",
    "Redis как backend для кеширования в Python",
    "asyncio и кеширование асинхронных функций",
]

response = co.rerank(
    model="rerank-multilingual-v3.0",   # или "rerank-english-v3.0"
    query=query,
    documents=candidates,
    top_n=3,                            # вернуть только топ-3
    return_documents=True,              # включить текст в ответ
)

print(f"Query: {query}\n")
for result in response.results:
    print(
        f"#{result.index + 1:2d}  "
        f"relevance={result.relevance_score:.3f}  "
        f"{result.document.text}"
    )

from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
import cohere, os

qdrant = QdrantClient(url="http://localhost:6333")
embed  = SentenceTransformer("all-MiniLM-L6-v2")
co     = cohere.Client(os.environ["COHERE_API_KEY"])


def rag_retrieve_rerank(
    query: str,
    collection: str,
    retrieve_n: int = 100,   # ANN: берём с запасом
    rerank_top_k: int = 5,   # Reranker: финальный топ для LLM
) -> list[dict]:
    """Двухэтапный retrieval: ANN overfetch → Cohere rerank."""

    # ── Этап 1: ANN-поиск с overfetch ─────────────────────────────────
    q_vec = embed.encode(query, normalize_embeddings=True).tolist()
    ann_results = qdrant.search(
        collection_name=collection,
        query_vector=q_vec,
        limit=retrieve_n,
        with_payload=True,
    )

    if not ann_results:
        return []

    # ── Этап 2: Cohere Rerank ──────────────────────────────────────────
    docs_for_rerank = [r.payload["text"] for r in ann_results]

    reranked = co.rerank(
        model="rerank-multilingual-v3.0",
        query=query,
        documents=docs_for_rerank,
        top_n=rerank_top_k,
        return_documents=True,
    )

    # Восстанавливаем полный payload по индексу кандидата
    final = []
    for r in reranked.results:
        original = ann_results[r.index]
        final.append({
            "text":            r.document.text,
            "relevance_score": r.relevance_score,
            "ann_score":       original.score,   # для сравнения / дебаггинга
            "payload":         original.payload,
        })

    return final


# Использование
results = rag_retrieve_rerank(
    query="обработка ошибок подключения Redis в Python",
    collection="knowledge_base",
    retrieve_n=100,
    rerank_top_k=5,
)

for i, doc in enumerate(results, 1):
    delta = "↑" if doc["relevance_score"] > 0.5 else "↓"
    print(
        f"#{i} {delta}  rel={doc['relevance_score']:.3f}  "
        f"ann={doc['ann_score']:.3f}  {doc['text'][:70]}"
    )

pip install sentence-transformers transformers torch

from sentence_transformers import CrossEncoder

# bge-reranker-v2-m3 поддерживает русский, китайский, английский и др.
bge_reranker = CrossEncoder(
    "BAAI/bge-reranker-v2-m3",
    max_length=512,          # max длина пары (query + doc tokens)
    device="cpu",            # или "cuda" при наличии GPU
)

query = "настройка nginx для SSL-терминации с Let's Encrypt"

candidates = [
    "Nginx: настройка HTTPS и SSL-сертификатов",
    "Let's Encrypt certbot: автоматическое получение сертификатов",
    "SSL-терминация в nginx: проксирование HTTPS на HTTP бэкенд",
    "Nginx как reverse proxy: конфигурация и балансировка",
    "Certbot и nginx: полная инструкция по настройке",
    "HTTP/2 в nginx: включение и оптимизация",
    "Настройка firewall ufw для nginx (порты 80 и 443)",
]

# BGE возвращает score в [0, 1] без дополнительной нормализации
scores = bge_reranker.predict([[query, doc] for doc in candidates])

ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
for rank, (doc, score) in enumerate(ranked, 1):
    print(f"#{rank}  [{score:.4f}]  {doc}")

from sentence_transformers import CrossEncoder
import numpy as np

class BGEReranker:
    """Production-ready BGE reranker с batching и GPU-поддержкой."""

    def __init__(
        self,
        model_name: str = "BAAI/bge-reranker-v2-m3",
        device: str = "cpu",
        max_length: int = 512,
        batch_size: int = 32,
    ):
        self.model = CrossEncoder(
            model_name,
            max_length=max_length,
            device=device,
        )
        self.batch_size = batch_size

    def rerank(
        self,
        query: str,
        candidates: list[str],
        top_k: int | None = None,
    ) -> list[tuple[str, float]]:
        """
        Ранжирует кандидатов, возвращает (doc, score) по убыванию.
        При top_k=None возвращает все.
        """
        if not candidates:
            return []

        pairs = [[query, doc] for doc in candidates]

        # show_progress_bar=False убирает tqdm в продакшене
        scores = self.model.predict(
            pairs,
            batch_size=self.batch_size,
            show_progress_bar=False,
        )

        ranked = sorted(
            zip(candidates, scores.tolist()),
            key=lambda x: x[1],
            reverse=True,
        )

        return ranked if top_k is None else ranked[:top_k]

    def score_pair(self, query: str, doc: str) -> float:
        """Скор одной пары (для дебаггинга или LLM-оценки)."""
        return float(self.model.predict([[query, doc]])[0])


# Использование
reranker = BGEReranker(device="cpu", batch_size=16)

results = reranker.rerank(
    query="настройка nginx SSL Let's Encrypt",
    candidates=candidates,
    top_k=3,
)

for doc, score in results:
    print(f"[{score:.4f}] {doc}")

from dataclasses import dataclass
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
from sentence_transformers import SentenceTransformer, CrossEncoder
from anthropic import Anthropic
import torch


@dataclass
class RAGConfig:
    # Retrieval
    retrieve_n: int = 100       # кандидатов от ANN (с запасом)
    rerank_top_k: int = 5       # финальных документов для LLM
    min_relevance: float = 0.3  # порог отсечения нерелевантных

    # Models
    embed_model: str = "all-MiniLM-L6-v2"
    rerank_model: str = "BAAI/bge-reranker-v2-m3"


class RAGPipeline:
    def __init__(self, qdrant_url: str, collection: str, cfg: RAGConfig):
        self.qdrant     = QdrantClient(url=qdrant_url)
        self.collection = collection
        self.cfg        = cfg

        self.embedder = SentenceTransformer(cfg.embed_model)
        self.reranker = CrossEncoder(cfg.rerank_model)
        self.llm      = Anthropic()

    def retrieve(self, query: str, filters: Filter | None = None) -> list[dict]:
        """Этап 1: ANN с overfetch."""
        q_vec = self.embedder.encode(query, normalize_embeddings=True).tolist()
        hits = self.qdrant.search(
            collection_name=self.collection,
            query_vector=q_vec,
            limit=self.cfg.retrieve_n,
            query_filter=filters,
            with_payload=True,
        )
        return [{"text": h.payload["text"], "meta": h.payload, "ann_score": h.score}
                for h in hits]

    def rerank(self, query: str, candidates: list[dict]) -> list[dict]:
        """Этап 2: Cross-encoder reranking."""
        if not candidates:
            return []

        texts = [c["text"] for c in candidates]
        pairs = [[query, t] for t in texts]

        scores = self.reranker.predict(pairs, activation_fct=torch.sigmoid)

        # Объединяем score с метаданными
        for cand, score in zip(candidates, scores):
            cand["rerank_score"] = float(score)

        # Сортируем по rerank_score, отсекаем по порогу
        ranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)
        filtered = [c for c in ranked if c["rerank_score"] >= self.cfg.min_relevance]
        return filtered[:self.cfg.rerank_top_k]

    def generate(self, query: str, context_docs: list[dict]) -> str:
        """Этап 3: LLM generation с контекстом."""
        context = "\n\n".join(
            f"[{i+1}] {doc['text']}" for i, doc in enumerate(context_docs)
        )
        response = self.llm.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": (
                    f"Ответь на вопрос, используя только приведённые источники.\n\n"
                    f"Источники:\n{context}\n\n"
                    f"Вопрос: {query}"
                ),
            }],
        )
        return response.content[0].text

    def query(
        self,
        question: str,
        tenant_id: str,
        user_role_level: int = 1,
    ) -> dict:
        """Полный RAG-цикл: retrieve → rerank → generate."""

        # Metadata filter: изоляция по тенанту + RBAC
        filters = Filter(must=[
            FieldCondition(key="tenant_id",    match=MatchValue(value=tenant_id)),
            FieldCondition(key="is_verified",  match=MatchValue(value=True)),
            FieldCondition(key="access_level", range=Range(lte=user_role_level)),
        ])

        # Шаг 1: ANN retrieval
        candidates = self.retrieve(question, filters=filters)

        # Шаг 2: Reranking
        ranked = self.rerank(question, candidates)

        # Шаг 3: Generation
        answer = self.generate(question, ranked)

        return {
            "answer":     answer,
            "sources":    ranked,
            "candidates": len(candidates),
            "used":       len(ranked),
        }


# ── Использование ──────────────────────────────────────────────────────
pipeline = RAGPipeline(
    qdrant_url="http://localhost:6333",
    collection="knowledge_base",
    cfg=RAGConfig(retrieve_n=100, rerank_top_k=5),
)

result = pipeline.query(
    question="как обработать ошибку ConnectionError при работе с Redis?",
    tenant_id="acme-corp",
    user_role_level=1,
)

print(f"Найдено кандидатов: {result['candidates']}")
print(f"Использовано после reranking: {result['used']}")
print(f"\nОтвет:\n{result['answer']}")
print("\nИсточники:")
for src in result["sources"]:
    print(f"  rerank={src['rerank_score']:.3f}  ann={src['ann_score']:.3f}  {src['text'][:60]}")