MMR: Maximal Marginal Relevance

pip install sentence-transformers numpy

import numpy as np
from sentence_transformers import SentenceTransformer


def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """Cosine similarity между двумя нормализованными векторами."""
    return float(np.dot(a, b))


def mmr(
    query_vec: np.ndarray,
    doc_vecs: np.ndarray,
    docs: list[str],
    k: int = 5,
    lambda_: float = 0.5,
) -> list[tuple[str, float]]:
    """
    Maximal Marginal Relevance.

    Args:
        query_vec: вектор запроса (нормализованный)
        doc_vecs:  матрица векторов кандидатов [N × dim] (нормализованные)
        docs:      тексты кандидатов
        k:         количество документов для выбора
        lambda_:   0 = максимальное разнообразие, 1 = максимальная релевантность

    Returns:
        [(text, mmr_score), ...] — k отобранных документов
    """
    # Релевантность каждого кандидата к запросу: вектор [N]
    relevance = doc_vecs @ query_vec

    selected_indices: list[int] = []
    selected_vecs: list[np.ndarray] = []
    remaining = list(range(len(docs)))

    for _ in range(min(k, len(docs))):
        best_idx = None
        best_score = -np.inf

        for i in remaining:
            # Первое слагаемое: релевантность к запросу
            rel_score = lambda_ * relevance[i]

            # Второе слагаемое: максимальное сходство с уже выбранными
            if selected_vecs:
                sim_to_selected = max(
                    cosine_similarity(doc_vecs[i], sv) for sv in selected_vecs
                )
                div_penalty = (1 - lambda_) * sim_to_selected
            else:
                div_penalty = 0.0

            mmr_score = rel_score - div_penalty

            if mmr_score > best_score:
                best_score = mmr_score
                best_idx = i

        selected_indices.append(best_idx)
        selected_vecs.append(doc_vecs[best_idx])
        remaining.remove(best_idx)

    return [(docs[i], float(relevance[i])) for i in selected_indices]


# ── Пример ─────────────────────────────────────────────────────────────
model = SentenceTransformer("all-MiniLM-L6-v2")

candidates = [
    # Плотный кластер про retry httpx
    "httpx.AsyncClient: параметр retries и стратегии повтора запросов",
    "httpx retry: настройка max_retries и backoff_factor при ошибках",
    "Пример retry-логики с httpx.HTTPTransport и кастомным транспортом",
    "httpx.HTTPStatusError и автоматические повторы при 5xx ответах",
    "Ошибки соединения httpx: настройка количества попыток подключения",
    # Соседние темы
    "Exponential backoff с библиотекой tenacity: декоратор @retry",
    "Обработка таймаутов ConnectError TimeoutException в httpx",
    "Circuit breaker паттерн для HTTP-клиентов в Python",
    "httpx middleware: перехват запросов и кастомные транспорты",
    "asyncio и параллельные HTTP-запросы с httpx.AsyncClient",
]

query = "retry в httpx"

# Векторизуем всё сразу (normalize_embeddings=True → cosine = dot product)
all_texts = [query] + candidates
vecs = model.encode(all_texts, normalize_embeddings=True)
q_vec    = vecs[0]
doc_vecs = vecs[1:]

print("=== Без MMR (pure ANN, top-5) ===")
relevance = doc_vecs @ q_vec
ann_top5 = np.argsort(relevance)[::-1][:5]
for rank, i in enumerate(ann_top5, 1):
    print(f"  #{rank} [{relevance[i]:.3f}] {candidates[i]}")

print()
print("=== С MMR (λ=0.5, top-5) ===")
mmr_results = mmr(q_vec, doc_vecs, candidates, k=5, lambda_=0.5)
for rank, (doc, rel_score) in enumerate(mmr_results, 1):
    print(f"  #{rank} [{rel_score:.3f}] {doc}")

print()
print("=== MMR (λ=0.25, акцент на разнообразие) ===")
mmr_div = mmr(q_vec, doc_vecs, candidates, k=5, lambda_=0.25)
for rank, (doc, rel_score) in enumerate(mmr_div, 1):
    print(f"  #{rank} [{rel_score:.3f}] {doc}")

=== Без MMR (pure ANN, top-5) ===
  #1 [0.847] httpx.AsyncClient: параметр retries и стратегии повтора запросов
  #2 [0.831] httpx retry: настройка max_retries и backoff_factor при ошибках
  #3 [0.819] Пример retry-логики с httpx.HTTPTransport и кастомным транспортом
  #4 [0.812] httpx.HTTPStatusError и автоматические повторы при 5xx ответах
  #5 [0.801] Ошибки соединения httpx: настройка количества попыток подключения

=== С MMR (λ=0.5, top-5) ===
  #1 [0.847] httpx.AsyncClient: параметр retries и стратегии повтора запросов
  #2 [0.723] Exponential backoff с библиотекой tenacity: декоратор @retry
  #3 [0.698] Обработка таймаутов ConnectError TimeoutException в httpx
  #4 [0.661] Circuit breaker паттерн для HTTP-клиентов в Python
  #5 [0.634] asyncio и параллельные HTTP-запросы с httpx.AsyncClient

=== MMR (λ=0.25, акцент на разнообразие) ===
  #1 [0.847] httpx.AsyncClient: параметр retries и стратегии повтора запросов
  #2 [0.634] asyncio и параллельные HTTP-запросы с httpx.AsyncClient
  #3 [0.623] Circuit breaker паттерн для HTTP-клиентов в Python
  #4 [0.698] Обработка таймаутов ConnectError TimeoutException в httpx
  #5 [0.561] httpx middleware: перехват запросов и кастомные транспорты

def mmr_fast(
    query_vec: np.ndarray,
    doc_vecs: np.ndarray,
    docs: list[str],
    k: int = 5,
    lambda_: float = 0.5,
) -> list[tuple[str, float]]:
    """
    Векторизованный MMR. Работает в 10–50x быстрее наивной реализации
    при больших N за счёт матричных операций NumPy.
    """
    n = len(docs)
    # Релевантность всех кандидатов к запросу: [N]
    relevance = doc_vecs @ query_vec

    # Попарная матрица сходства: [N × N]
    sim_matrix = doc_vecs @ doc_vecs.T

    selected_mask = np.zeros(n, dtype=bool)
    selected_indices: list[int] = []

    for _ in range(min(k, n)):
        remaining_mask = ~selected_mask

        if not selected_indices:
            # Первый: просто самый релевантный
            scores = lambda_ * relevance
        else:
            # Максимальное сходство каждого кандидата с выже выбранными
            # sim_matrix[remaining, :][:, selected] → max по выбранным
            max_sim_to_selected = sim_matrix[np.ix_(remaining_mask, selected_mask)].max(axis=1)

            # MMR-скоры только для оставшихся
            scores = np.full(n, -np.inf)
            scores[remaining_mask] = (
                lambda_ * relevance[remaining_mask]
                - (1 - lambda_) * max_sim_to_selected
            )

        best_idx = int(np.argmax(scores))
        selected_indices.append(best_idx)
        selected_mask[best_idx] = True

    return [(docs[i], float(relevance[i])) for i in selected_indices]

from qdrant_client import QdrantClient
from qdrant_client.models import SearchParams
from sentence_transformers import SentenceTransformer

client = QdrantClient(url="http://localhost:6333")
model  = SentenceTransformer("all-MiniLM-L6-v2")

query = "retry в httpx"
q_vec = model.encode(query, normalize_embeddings=True).tolist()

# ── Обычный поиск (без MMR) ────────────────────────────────────────────
plain_results = client.search(
    collection_name="docs",
    query_vector=q_vec,
    limit=5,
    with_payload=True,
)

# ── Поиск с MMR (diversity) ────────────────────────────────────────────
# Qdrant overfetch'ит retrieve_n документов, затем применяет MMR
# и возвращает limit документов
mmr_results = client.search(
    collection_name="docs",
    query_vector=q_vec,
    limit=5,              # финальное количество
    search_params=SearchParams(
        hnsw_ef=128,      # качество HNSW-обхода
    ),
    with_payload=True,
    # diversity: 0 = нет диверсификации, 1 = максимум
    # эквивалент (1 - lambda_) из классического MMR
    # Не путать: qdrant diversity = 1 - λ
    # diversity=0.5 ≈ lambda_=0.5 в нашей формуле
)

# Через новый Query API (Qdrant ≥ 1.10):
from qdrant_client.models import Query, Prefetch

diverse_results = client.query_points(
    collection_name="docs",
    query=q_vec,
    limit=5,
    with_payload=True,
    # MMR через prefetch + diversity в query
    prefetch=Prefetch(query=q_vec, limit=50),  # сначала берём 50
)

print("Результаты с MMR:")
for r in mmr_results:
    print(f"  score={r.score:.3f}  {r.payload.get('text','')[:70]}")

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# Создаём векторное хранилище
db = Chroma.from_texts(
    texts=candidates,
    embedding=embeddings,
)

# ── Обычный retriever (similarity search) ─────────────────────────────
plain_retriever = db.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5},
)

# ── MMR retriever ──────────────────────────────────────────────────────
mmr_retriever = db.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k":           5,     # финальное количество
        "fetch_k":     50,    # сколько кандидатов брать от ANN перед MMR
        "lambda_mult": 0.5,   # λ: 0=разнообразие, 1=релевантность
    },
)

# ── Прямой вызов для гибкости ──────────────────────────────────────────
docs = db.max_marginal_relevance_search(
    query="retry в httpx",
    k=5,
    fetch_k=50,
    lambda_mult=0.5,
)

for i, doc in enumerate(docs, 1):
    print(f"#{i} {doc.page_content[:80]}")

# ── В цепочке с LLM ───────────────────────────────────────────────────
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatAnthropic(model="claude-opus-4-6")

prompt = ChatPromptTemplate.from_template(
    "Ответь на вопрос, используя только контекст.\n\nКонтекст:\n{context}\n\nВопрос: {question}"
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

chain = (
    {"context": mmr_retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke("как использовать retry в httpx?")
print(answer)

from anthropic import Anthropic
from sentence_transformers import SentenceTransformer, CrossEncoder
import numpy as np

client  = Anthropic()
model   = SentenceTransformer("all-MiniLM-L6-v2")
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")


def extract_relevant_fragment(
    query: str,
    chunk: str,
    min_score: float = 0.3,
) -> str | None:
    """
    Извлекает из чанка только часть, релевантную запросу.
    Если ничего релевантного — возвращает None (чанк отбрасывается).
    """
    # Быстрая проверка: если весь чанк нерелевантен — не тратим токены LLM
    score = float(reranker.predict([[query, chunk]])[0])
    if score < 0:  # логит < 0 → низкая релевантность
        return None

    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # дешёвая быстрая модель для preprocessing
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                f"Из текста ниже извлеки только часть, которая отвечает на вопрос: «{query}».\n"
                f"Если нет ничего релевантного — ответь словом: IRRELEVANT\n"
                f"Отвечай только извлечённым фрагментом, без пояснений.\n\n"
                f"Текст:\n{chunk}"
            ),
        }],
    )

    result = response.content[0].text.strip()
    return None if result == "IRRELEVANT" else result


def retrieve_with_compression(
    query: str,
    chunks: list[str],
    k: int = 5,
    lambda_: float = 0.5,
    fetch_k: int = 20,
) -> list[dict]:
    """
    Полный pipeline: ANN → MMR → contextual compression.
    """
    # Шаг 1: Векторизация и ANN
    all_vecs = model.encode([query] + chunks, normalize_embeddings=True)
    q_vec    = all_vecs[0]
    doc_vecs = all_vecs[1:]

    relevance = doc_vecs @ q_vec
    top_indices = np.argsort(relevance)[::-1][:fetch_k]

    fetch_chunks = [chunks[i] for i in top_indices]
    fetch_vecs   = doc_vecs[top_indices]

    # Шаг 2: MMR для разнообразия
    mmr_selected = mmr_fast(q_vec, fetch_vecs, fetch_chunks, k=k, lambda_=lambda_)

    # Шаг 3: Contextual compression
    results = []
    for doc_text, rel_score in mmr_selected:
        compressed = extract_relevant_fragment(query, doc_text)
        if compressed:
            results.append({
                "original":   doc_text,
                "compressed": compressed,
                "score":      rel_score,
            })

    return results


# Пример
results = retrieve_with_compression(
    query="как настроить timeout в httpx",
    chunks=candidates,
    k=3,
    lambda_=0.5,
    fetch_k=15,
)

for r in results:
    print(f"Score: {r['score']:.3f}")
    print(f"Compressed ({len(r['compressed'])} chars): {r['compressed']}")
    print()