Query Transformation: уточнение и переформулировка запроса

from anthropic import Anthropic

client = Anthropic()

REWRITE_PROMPT = """\
Переформулируй запрос пользователя для поиска в технической документации.

Правила:
- Сделай запрос явным и самодостаточным — без местоимений «это», «оно», «там»
- Добавь технические термины, если они подразумеваются
- Сохрани исходный смысл, не расширяй тему
- Верни только переформулированный запрос, без пояснений

Запрос: {query}"""


def rewrite_query(query: str, model: str = "claude-haiku-4-5-20251001") -> str:
    """
    Переформулирует запрос для улучшения retrieval.
    Использует дешёвую модель — это не финальная генерация, а preprocessing.
    """
    response = client.messages.create(
        model=model,
        max_tokens=128,
        messages=[{
            "role": "user",
            "content": REWRITE_PROMPT.format(query=query),
        }],
    )
    return response.content[0].text.strip()


# ── Примеры ─────────────────────────────────────────────────────────
examples = [
    "не работает таймаут",
    "как сделать быстрее",
    "а у asyncclient так же?",
    "ошибка 429 что делать",
]

for q in examples:
    rewritten = rewrite_query(q)
    print(f"Было:  {q}")
    print(f"Стало: {rewritten}")
    print()

# Было:  не работает таймаут
# Стало: httpx ConnectTimeout exception: настройка и обработка таймаута
#
# Было:  как сделать быстрее
# Стало: оптимизация производительности HTTP-запросов в httpx: connection pooling, keep-alive
#
# Было:  а у asyncclient так же?
# Стало: настройка таймаутов в httpx AsyncClient — совпадает ли с синхронным Client?
#
# Было:  ошибка 429 что делать
# Стало: httpx обработка HTTP 429 Too Many Requests: retry-стратегия и rate limiting

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

MULTI_QUERY_PROMPT = """\
Сгенерируй {n} разных формулировок запроса для поиска в документации.
Каждая формулировка должна подходить к теме с другого угла: другие термины,
другой уровень детализации, другой тип вопроса (как?, почему?, что такое?).

Верни только список формулировок — по одной в строке, без нумерации.

Исходный запрос: {query}"""


async def generate_queries(
    query: str,
    n: int = 4,
    model: str = "claude-haiku-4-5-20251001",
) -> list[str]:
    """Генерирует N семантически разных вариантов запроса."""
    response = await client.messages.create(
        model=model,
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": MULTI_QUERY_PROMPT.format(query=query, n=n),
        }],
    )
    lines = response.content[0].text.strip().splitlines()
    queries = [q.strip().lstrip("•-–").strip() for q in lines if q.strip()]
    return [query] + queries[:n]  # добавляем исходный запрос


async def multi_query_retrieve(
    query: str,
    retriever,          # callable: query_str -> list[RetrievedChunk]
    n_variants: int = 3,
    top_k_per_query: int = 3,
) -> list:
    """
    Retrieval по N формулировкам с дедупликацией по chunk_id.
    """
    queries = await generate_queries(query, n=n_variants)

    # Параллельный retrieval по всем формулировкам
    tasks = [retriever(q, top_k=top_k_per_query) for q in queries]
    results_per_query = await asyncio.gather(*tasks)

    # Дедупликация: если один чанк нашёлся несколько раз — берём с лучшим score
    seen: dict[str, object] = {}
    for chunks in results_per_query:
        for chunk in chunks:
            chunk_id = chunk.chunk_id
            if chunk_id not in seen or chunk.score > seen[chunk_id].score:
                seen[chunk_id] = chunk

    # Сортируем по score
    return sorted(seen.values(), key=lambda c: c.score, reverse=True)


# ── Пример ─────────────────────────────────────────────────────────
async def main():
    query = "не работает таймаут httpx"
    queries = await generate_queries(query, n=4)
    for i, q in enumerate(queries):
        print(f"[{i}] {q}")

# [0] не работает таймаут httpx  ← исходный
# [1] httpx.Timeout: настройка connect и read таймаутов
# [2] ConnectTimeout exception в httpx: причины и обработка
# [3] httpx клиент зависает при подключении: как установить лимит времени
# [4] параметры таймаута httpx: connect, read, write, pool

from anthropic import Anthropic
from sentence_transformers import SentenceTransformer

client = Anthropic()
embed_model = SentenceTransformer("BAAI/bge-small-en-v1.5")

HYDE_PROMPT = """\
Напиши короткий (2–4 предложения) технический ответ на вопрос, как будто ты пишешь документацию.
Используй технические термины и конкретные примеры API.
Ответ может быть неточным — главное, чтобы он был сформулирован как документация.
Без заголовков, только текст.

Вопрос: {query}"""


def generate_hypothetical_document(query: str) -> str:
    """Генерирует гипотетический документ для HyDE."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=200,
        messages=[{"role": "user", "content": HYDE_PROMPT.format(query=query)}],
    )
    return response.content[0].text.strip()


def hyde_retrieve(
    query: str,
    qdrant_client,
    collection_name: str,
    top_k: int = 5,
) -> list:
    """
    Поиск через HyDE: генерируем гипотетический документ,
    ищем по его embedding.
    """
    # Шаг 1: Гипотетический ответ
    hyp_doc = generate_hypothetical_document(query)

    # Шаг 2: Embedding гипотетического документа
    hyp_vec = embed_model.encode(
        [hyp_doc], normalize_embeddings=True
    )[0].tolist()

    # Шаг 3: Поиск по этому вектору
    hits = qdrant_client.search(
        collection_name=collection_name,
        query_vector=hyp_vec,
        limit=top_k,
        with_payload=True,
    )
    return hits


# ── Пример гипотетического документа ───────────────────────────────
query = "как настроить таймаут в httpx"
hyp = generate_hypothetical_document(query)
print(hyp)
# → "Для управления таймаутами в httpx используйте класс httpx.Timeout,
#    который принимает отдельные параметры для каждого этапа: connect, read,
#    write и pool. Передайте экземпляр в Client(timeout=...) или задайте
#    per-request через client.get(url, timeout=5.0)."

import json
from anthropic import Anthropic

client = Anthropic()

DECOMPOSE_PROMPT = """\
Разбей сложный вопрос на простые подзапросы для поиска в документации.
Каждый подзапрос должен быть самодостаточным и отвечать на одну конкретную вещь.

Верни JSON-массив строк. Пример: ["подзапрос 1", "подзапрос 2"]
Максимум 4 подзапроса. Если вопрос уже простой — верни ["исходный вопрос"].

Вопрос: {query}"""

SYNTHESIZE_PROMPT = """\
На основе приведённых ответов на подзапросы дай исчерпывающий ответ на исходный вопрос.
Используй информацию из всех подответов. Ответ должен быть структурированным и логичным.

Исходный вопрос: {question}

Ответы на подзапросы:
{sub_answers}"""


def decompose_query(query: str) -> list[str]:
    """Разбивает сложный вопрос на подзапросы."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=256,
        messages=[{"role": "user",
                   "content": DECOMPOSE_PROMPT.format(query=query)}],
    )
    try:
        text = response.content[0].text.strip()
        # Ищем JSON-массив в ответе
        start = text.find("[")
        end   = text.rfind("]") + 1
        return json.loads(text[start:end])
    except Exception:
        return [query]  # fallback: исходный запрос


def decomposed_rag(
    query: str,
    rag_pipeline,    # объект с методом .generate(query) -> str
) -> str:
    """
    Разбивает запрос, отвечает на каждый подзапрос, синтезирует итог.
    """
    sub_queries = decompose_query(query)

    if len(sub_queries) == 1:
        # Простой вопрос — обычный RAG
        return rag_pipeline.generate(sub_queries[0]).answer

    # Ответы на каждый подзапрос
    sub_answers = []
    for sq in sub_queries:
        result = rag_pipeline.generate(sq)
        if not result.skipped:
            sub_answers.append(f"Вопрос: {sq}\nОтвет: {result.answer}")

    if not sub_answers:
        return "Информация по данному вопросу не найдена в документации."

    # Синтез финального ответа
    synthesis = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": SYNTHESIZE_PROMPT.format(
                question=query,
                sub_answers="\n\n---\n\n".join(sub_answers),
            ),
        }],
    )
    return synthesis.content[0].text


# ── Пример ─────────────────────────────────────────────────────────
query = "в чём разница между httpx и requests и как мигрировать?"
sub_queries = decompose_query(query)
print(sub_queries)
# → ["сравнение httpx и requests: функции и производительность",
#    "API httpx vs requests: ключевые отличия",
#    "миграция с requests на httpx: пошаговое руководство"]

from anthropic import Anthropic

client = Anthropic()

STEP_BACK_PROMPT = """\
Сформулируй более общий (step-back) вопрос, который охватывает тему данного вопроса.
Цель: найти концептуальную информацию, которая поможет ответить на конкретный вопрос.

Конкретный вопрос: {query}
Общий вопрос (1 строка, без пояснений):"""


def step_back_query(query: str) -> str:
    """Генерирует обобщённый вариант запроса."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=80,
        messages=[{"role": "user",
                   "content": STEP_BACK_PROMPT.format(query=query)}],
    )
    return response.content[0].text.strip()


def step_back_retrieve(
    query: str,
    retriever,
    top_k: int = 3,
) -> list:
    """
    Retrieval по двум запросам: конкретному и обобщённому.
    Результаты объединяются с дедупликацией.
    """
    # Оба запроса
    general_query = step_back_query(query)

    specific_chunks = retriever(query,       top_k=top_k)
    general_chunks  = retriever(general_query, top_k=top_k)

    # Дедупликация: конкретные документы приоритетнее
    seen: dict[str, object] = {}
    for chunk in specific_chunks + general_chunks:
        if chunk.chunk_id not in seen:
            seen[chunk.chunk_id] = chunk

    return list(seen.values())


# ── Пример ─────────────────────────────────────────────────────────
q = "как передать timeout только для одного запроса в httpx?"
step_back = step_back_query(q)
print(f"Исходный:    {q}")
print(f"Step-back:   {step_back}")
# Step-back: «как работает механизм таймаутов в httpx — уровни Client и Request?»

import asyncio
from collections import defaultdict
from anthropic import AsyncAnthropic

client = AsyncAnthropic()


def reciprocal_rank_fusion(
    results_per_query: list[list],
    k: int = 60,
) -> list:
    """
    Объединяет ранжированные списки через RRF.

    results_per_query: список списков чанков,
                       каждый список отсортирован по score (лучший первый).
    k: константа сглаживания (обычно 60)
    """
    scores: dict[str, float] = defaultdict(float)
    chunks_by_id: dict[str, object] = {}

    for ranked_list in results_per_query:
        for rank, chunk in enumerate(ranked_list, start=1):
            chunk_id = chunk.chunk_id
            scores[chunk_id] += 1.0 / (k + rank)
            # Сохраняем объект чанка (первое вхождение)
            if chunk_id not in chunks_by_id:
                chunks_by_id[chunk_id] = chunk

    # Сортируем по RRF score
    sorted_ids = sorted(scores, key=lambda cid: scores[cid], reverse=True)
    result = []
    for cid in sorted_ids:
        chunk = chunks_by_id[cid]
        chunk.rrf_score = scores[cid]  # добавляем RRF score для отладки
        result.append(chunk)

    return result


async def rag_fusion_retrieve(
    query: str,
    retriever,            # async callable: (query, top_k) -> list[Chunk]
    n_variants: int = 3,
    top_k_per_query: int = 5,
) -> list:
    """
    Полный RAG-Fusion pipeline:
    1. Генерация N вариантов запроса
    2. Параллельный retrieval по каждому
    3. RRF-объединение результатов
    """
    # Генерация вариантов запроса
    queries = await generate_queries(query, n=n_variants)

    # Параллельный retrieval
    tasks = [retriever(q, top_k=top_k_per_query) for q in queries]
    all_results = await asyncio.gather(*tasks)

    # RRF fusion
    fused = reciprocal_rank_fusion(list(all_results))

    # Возвращаем топ-5 по RRF
    return fused[:5]


# ── Пример ─────────────────────────────────────────────────────────
async def main():
    query = "как настроить таймауты в httpx"
    chunks = await rag_fusion_retrieve(query, retriever=my_retriever)
    for c in chunks:
        print(f"RRF {c.rrf_score:.4f} | {c.source} | {c.text[:60]}...")

asyncio.run(main())

import re
from dataclasses import dataclass
from enum import Enum


class Strategy(Enum):
    SIMPLE    = "simple"      # только rewriting
    MULTI     = "multi"       # multi-query
    HYDE      = "hyde"        # HyDE
    DECOMPOSE = "decompose"   # decomposition


def classify_query(query: str) -> Strategy:
    """
    Простая эвристика для выбора стратегии.
    В продакшне можно заменить на LLM-классификатор.
    """
    q = query.lower()

    # Составные / сравнительные → декомпозиция
    compound_signals = [" и как ", " и почему ", "чем отличается", "сравни ", "разница между"]
    if any(s in q for s in compound_signals):
        return Strategy.DECOMPOSE

    # Короткие (< 4 слов) → multi-query для расширения охвата
    if len(q.split()) < 4:
        return Strategy.MULTI

    # Конкретные технические вопросы → HyDE
    tech_signals = [".py", "class ", "method", "параметр", "аргумент", "exception"]
    if any(s in q for s in tech_signals):
        return Strategy.HYDE

    # По умолчанию — простой rewriting
    return Strategy.SIMPLE


@dataclass
class TransformResult:
    strategy:  Strategy
    queries:   list[str]   # один или несколько запросов для retrieval
    original:  str


async def transform_query(
    query: str,
    strategy: Strategy | None = None,
) -> TransformResult:
    """
    Трансформирует запрос по выбранной (или автоопределённой) стратегии.
    """
    if strategy is None:
        strategy = classify_query(query)

    if strategy == Strategy.SIMPLE:
        rewritten = rewrite_query(query)
        return TransformResult(strategy, [rewritten], query)

    elif strategy == Strategy.MULTI:
        queries = await generate_queries(query, n=3)
        return TransformResult(strategy, queries, query)

    elif strategy == Strategy.HYDE:
        hyp_doc = generate_hypothetical_document(query)
        # Для HyDE возвращаем гипотетический документ как «запрос»
        return TransformResult(strategy, [hyp_doc], query)

    elif strategy == Strategy.DECOMPOSE:
        sub_queries = decompose_query(query)
        return TransformResult(strategy, sub_queries, query)

    return TransformResult(Strategy.SIMPLE, [query], query)


# ── Использование ───────────────────────────────────────────────────
async def main():
    queries = [
        "не работает таймаут",                                   # → MULTI
        "как передать httpx.Timeout в AsyncClient.get() method", # → HYDE
        "чем httpx отличается от requests и как мигрировать?",   # → DECOMPOSE
        "как настроить таймаут соединения в httpx",              # → SIMPLE
    ]

    for q in queries:
        result = await transform_query(q)
        print(f"[{result.strategy.value}] {q!r}")
        for tq in result.queries:
            print(f"  → {tq!r}")
        print()

asyncio.run(main())