Улучшение поисковой выдачи ассистента: группировка и склейка чанков, снижение повторов, повышение разнообразия источников, простая re-ranking модель и схемы интеграции поиска с ассистентом.
То, что сейчас возвращает rag_retrieve:
[ { "text": "Фрагмент №0 страницы A", "page_id": 1, "chunk_index": 0, "score": 42 }, { "text": "Фрагмент №1 страницы A", "page_id": 1, "chunk_index": 1, "score": 40 }, { "text": "Фрагмент страницы B", "page_id": 2, "chunk_index": 0, "score": 39 } ]
Типичные проблемы:
Задача - решить это без истерики и без переусложнения.
Сначала сгруппируем результаты по page_id, а внутри отсортируем по chunk_index.
Допишем хелпер поверх rag_retrieve:
# rag_postprocess.py from collections import defaultdict from typing import List, Dict, Any def group_by_page(chunks: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Принимает список чанков (как возвращает rag_retrieve), отдаёт список страниц с их чанками. Формат: [ { "page_id": ..., "slug": ..., "title": ..., "url": ... (если есть), "chunks": [ {chunk...}, ... ], "best_score": ..., }, ... ] """ pages = defaultdict(list) for ch in chunks: pages[ch["page_id"]].append(ch) grouped = [] for page_id, ch_list in pages.items(): ch_list_sorted = sorted(ch_list, key=lambda c: c.get("chunk_index", 0)) # берём максимум score по странице best_score = max(c.get("score") or 0 for c in ch_list_sorted) grouped.append({ "page_id": page_id, "slug": ch_list_sorted[0].get("slug"), "title": ch_list_sorted[0].get("title"), "url": ch_list_sorted[0].get("url"), # если мы добавляли URL раньше "chunks": ch_list_sorted, "best_score": best_score, }) # отсортируем страницы по лучшему скору grouped.sort(key=lambda p: p["best_score"], reverse=True) return grouped
Дальше будем работать не с "плоским" списком, а с страницами.
Обычно мы хотим дать ассистенту целостный кусок, а не "абзац через один". Простая стратегия:
chunk_index;def merge_adjacent_chunks(chunks: List[Dict[str, Any]], max_cluster_gap: int = 1) -> List[str]: """ chunks должны быть отсортированы по chunk_index. Возвращает список текстовых блоков (кластеров). """ if not chunks: return [] clusters: List[List[str]] = [] current_cluster: List[str] = [chunks[0]["text"]] last_index = chunks[0].get("chunk_index", 0) for ch in chunks[1:]: idx = ch.get("chunk_index", last_index) # если чанк рядом с предыдущим — добавляем в тот же кластер if idx - last_index <= max_cluster_gap: current_cluster.append(ch["text"]) else: clusters.append(current_cluster) current_cluster = [ch["text"]] last_index = idx clusters.append(current_cluster) merged_blocks = ["\n\n".join(c) for c in clusters] return merged_blocks
Теперь можно сделать функцию, которая на уровне страниц строит уже "готовые блоки" контекста.
build_rag_contextsСоберём всё воедино: берём топ страниц, внутри склеиваем чанки, ограничиваем общий объём.
from typing import List, Dict, Any, Optional from rag_postprocess import group_by_page, merge_adjacent_chunks def build_rag_contexts( chunks: List[Dict[str, Any]], *, max_pages: int = 4, max_blocks: int = 6, max_chars_total: int = 8000, ) -> List[Dict[str, Any]]: """ На вход: "сырые" чанки из rag_retrieve (с полями text/title/slug/page_id/score/url/chunk_index). На выход: список контекстных блоков для LLM. Каждый блок: { "page_id": ..., "title": ..., "url": ..., "text": "...склеенный текст...", "score": ..., } """ pages = group_by_page(chunks) contexts: List[Dict[str, Any]] = [] total_chars = 0 for page in pages[:max_pages]: clusters = merge_adjacent_chunks(page["chunks"]) for block_text in clusters: if not block_text.strip(): continue if len(contexts) >= max_blocks: return contexts if total_chars + len(block_text) > max_chars_total: return contexts contexts.append({ "page_id": page["page_id"], "title": page["title"], "url": page.get("url"), "text": block_text, "score": page["best_score"], }) total_chars += len(block_text) return contexts
Теперь вместо "сырых" чанков ассистент получает:
Даже без отдельной re-ranking-модели можно немного улучшить порядок:
Наказание за "перепредставленные" страницы. Если у нас много чанков с одной страницы, мы уже сгруппировали и склеили, но можно чуть уменьшить score, если страница дала слишком много чанков.
Бонус за разнообразие тегов. Если у страницы редкий тег, который явно совпадает с запросом (например, docker, а в запросе есть слово "docker"), можно добавить к score.
Бонус за свежесть. Если в page_chunks есть published_at, можно:
boosted_score = base_score + freshness_boost
где freshness_boost зависит от давности.
import math from typing import List, Dict, Any def rerank_pages( pages: List[Dict[str, Any]], query: str, ) -> List[Dict[str, Any]]: """ На вход: результат group_by_page (страницы с best_score, title, tags?). На выход: тот же список, но отсортированный с учётом простых поправок. """ q = query.lower() def score_page(p: Dict[str, Any]) -> float: base = p["best_score"] or 0 title = (p.get("title") or "").lower() tags = [t.lower() for t in (p.get("tags") or [])] bonus = 0.0 # бонус, если запрос напрямую встречается в заголовке if q in title: bonus += 5.0 # бонус за совпадение по словам запроса в тегах for word in q.split(): if word in tags: bonus += 3.0 # лёгкое «сплющивание» очень больших скорингов, чтобы # отрыв лидера не был космическим return math.log1p(base) + bonus pages_sorted = sorted(pages, key=score_page, reverse=True) return pages_sorted
Это не научный re-ranking, но:
При желании можно заменить на LLM- или ML re-ranker позже.
rag_retrieve_v2 с пост-обработкойСклеим всё в более удобную функцию для ассистентного слоя - один вызов, сразу готовые блоки.
# rag_pipeline.py from typing import List, Dict, Any from rag_retrieve import rag_retrieve from rag_postprocess import group_by_page, merge_adjacent_chunks from rag_postprocess import build_rag_contexts # как выше from rag_postprocess import rerank_pages # если вынесешь отдельно def rag_retrieve_v2( query: str, *, limit_blocks: int = 6, max_chars: int = 8000, ) -> List[Dict[str, Any]]: # 1. Сырые чанки из Typesense raw_chunks = rag_retrieve(query, top_k=limit_blocks * 4, vector_k=limit_blocks * 8) if not raw_chunks: return [] # 2. Группируем по страницам pages = group_by_page(raw_chunks) # (опционально) 3. Re-rank страниц pages = rerank_pages(pages, query) # 4. Собираем блоки контекста # reuse build_rag_contexts, но слегка адаптируем, # чтобы он принимал уже rerank’нутые pages contexts: List[Dict[str, Any]] = [] total_chars = 0 for page in pages: clusters = merge_adjacent_chunks(page["chunks"]) for block_text in clusters: if not block_text.strip(): continue if len(contexts) >= limit_blocks: return contexts if total_chars + len(block_text) > max_chars: return contexts contexts.append({ "page_id": page["page_id"], "title": page["title"], "url": page.get("url"), "text": block_text, "score": page["best_score"], }) total_chars += len(block_text) return contexts
Теперь фронтенд / ассистентный код может дергать именно rag_retrieve_v2.
Заменяем /api/rag-search, чтобы он возвращал уже пост-обработанные блоки:
# app_rag.py (обновлённый) from flask import Flask, request, jsonify from rag_pipeline import rag_retrieve_v2 app = Flask(__name__) @app.route("/api/rag-search", methods=["POST"]) def api_rag_search(): data = request.get_json(force=True) or {} query = (data.get("query") or "").strip() limit = data.get("limit") or 6 try: limit = int(limit) except (TypeError, ValueError): limit = 6 limit = max(1, min(limit, 10)) if not query: return jsonify({"contexts": []}) contexts = rag_retrieve_v2(query, limit_blocks=limit, max_chars=8000) return jsonify({"contexts": contexts})
На стороне ассистента можно оставить тот же формат промпта, просто блоки теперь более чистые и разнообразные.
POST /api/rag-search => вызов LLM.Разделяем:
rag_retrieve_v2, складывает задачу в очередь (Redis, RabbitMQ).Это уже архитектура "ассистента как сервиса". RAG-слой (Typesense + код) в этой схеме:
Если ассистент только "для сайта":
JS на фронте:
POST /api/rag-search;Тут важно:
Чтобы не делать вслепую:
search_log можно добавлять source='rag':top_k, vector_k;