Merge upstream main and resolve conflicts

Reviewed-on: zovos/vk_hackathon#2

.ai_update/changes.md

@@ -23,13 +23,37 @@
 ## Договоренности
 
 - Не менять контракты `POST /index`, `POST /sparse_embedding`, `POST /search`.
-- По текущей задаче фокус держать на `index`.
-- Не трогать `search`, `docker-compose.yml` и инфраструктуру без отдельной просьбы.
+- По текущей задаче фокус держать на `index`, если пользователь не просит иначе.
+- Не трогать `docker-compose.yml` и инфраструктуру без отдельной просьбы.
 - Не откатывать чужие или пользовательские изменения.
 
 ## Записи
 
-### 2026-04-18 - индексный P1 и безопасный P0 search
+### 2026-04-18 - разрешены конфликты с upstream/main
+
+Что сделано:
+
+- Объединены записи журнала из ветки PR и `upstream/main`.
+- В `search/main.py` сохранены расширенные retrieval/rerank доработки из обеих веток.
+- Убраны конфликтные маркеры после merge `upstream/main`.
+- Сохранены безопасные лимиты retrieval и защита от падения rerank.
+
+Измененные файлы:
+
+- `.ai_update/changes.md`
+- `search/main.py`
+
+Проверки:
+
+- `python3 -m py_compile index/main.py search/main.py`
+- Прямой smoke `build_chunks` на `data/Go Nova.json`: 15 чанков, покрыто 25 из 25 сообщений.
+- Pure smoke для `search/main.py` helper-функций через stub-модули.
+
+Открытые риски:
+
+- Полный интеграционный прогон с настоящими Qdrant/dense/rerank зависит от внешних сервисов.
+
+### 2026-04-18 - index P1 and search retrieval/rerank
 
 Что сделано:
 
@@ -46,7 +70,8 @@
 - `question.variants`, `question.hyde`, `question.keywords` и entities используются как дополнительные dense/sparse запросы.
 - Retrieval делает несколько prefetch-запросов и fusion через Qdrant.
 - Rerank больше не выбрасывает retrieval-хвост.
-- Финальные `message_ids` агрегируются, дедуплицируются и ограничиваются `top-50`.
+- Retrieval points дедуплицируются по Qdrant point id.
+- Финальные `message_ids` агрегируются по score, дедуплицируются и ограничиваются `top-50`.
 
 Измененные файлы:
 
@@ -86,10 +111,10 @@
 
 Контекст по текущему состоянию:
 
-- `index/main.py` сейчас использует символьный chunking.
-- `render_message` берет только `message.text` и `parts[*].text`.
-- `page_content`, `dense_content`, `sparse_content` сейчас одинаковые.
-- В примере `data/Go Nova.json` текущий `build_chunks` покрывает 24 из 25 сообщений; системное сообщение с `member_event` выпадает из индекса.
+- `index/main.py` тогда использовал символьный chunking.
+- `render_message` брал только `message.text` и `parts[*].text`.
+- `page_content`, `dense_content`, `sparse_content` были одинаковые.
+- В примере `data/Go Nova.json` старый `build_chunks` покрывал 24 из 25 сообщений; системное сообщение с `member_event` выпадало из индекса.
 
 Измененные файлы:
 
@@ -98,7 +123,3 @@
 Проверки:
 
 - Код не запускался, потому что создан только журнал.
-
-Следующий ожидаемый фокус:
-
-- Перестроить индексный renderer и chunking в `index/main.py`, если пользователь попросит перейти к реализации.

search/main.py

@@ -179,6 +179,7 @@ FINAL_TOP_K = 50
 MAX_DENSE_QUERIES = 4
 MAX_SPARSE_QUERIES = 3
 
+
 async def embed_dense(client: httpx.AsyncClient, text: str) -> list[float]:
     # Dense endpoint ожидает OpenAI-compatible body с input как списком строк.
     response = await client.post(
@@ -277,11 +278,12 @@ def build_search_filter(question: Question) -> models.Filter | None:
 
     return models.Filter(must=must_conditions) if must_conditions else None
 
+
 async def qdrant_search(
     client: AsyncQdrantClient,
     dense_vectors: list[list[float]],
     sparse_vectors: list[SparseVector],
-    question_data: Question
+    question_data: Question,
 ) -> Any | None:
     search_filter = build_search_filter(question_data)
     prefetch: list[models.Prefetch] = []
@@ -329,6 +331,32 @@ async def qdrant_search(
     return response.points
 
 
+def deduplicate_points(points: list[Any]) -> list[Any]:
+    unique_points: list[Any] = []
+    seen_ids: set[str] = set()
+
+    for point in points:
+        point_id = getattr(point, "id", None)
+        if point_id is None:
+            unique_points.append(point)
+            continue
+
+        point_id = str(point_id)
+        if point_id in seen_ids:
+            continue
+        seen_ids.add(point_id)
+        unique_points.append(point)
+
+    return unique_points
+
+
+def extract_point_score(point: Any) -> float:
+    score = getattr(point, "score", 0.0)
+    if score is None:
+        return 0.0
+    return float(score)
+
+
 def extract_message_ids(point: Any) -> list[str]:
     payload = point.payload or {}
     metadata = payload.get("metadata") or {}
@@ -368,9 +396,9 @@ async def rerank_points(
     client: httpx.AsyncClient,
     query: str,
     points: list[Any],
-) -> list[Any]:
+) -> list[tuple[Any, float]]:
     rerank_candidates = points[:RERANK_LIMIT]
-    retrieval_tail = points[RERANK_LIMIT:]
+    tail_candidates = points[RERANK_LIMIT:]
     rerank_targets = [
         str((point.payload or {}).get("page_content") or "")
         for point in rerank_candidates
@@ -380,7 +408,7 @@ async def rerank_points(
         scores = await get_rerank_scores(client, query, rerank_targets)
     except Exception:
         logger.exception("Rerank failed, returning retrieval order")
-        return points
+        return [(point, extract_point_score(point)) for point in points]
 
     if len(scores) != len(rerank_candidates):
         logger.warning(
@@ -388,42 +416,49 @@ async def rerank_points(
             len(scores),
             len(rerank_candidates),
         )
-        return points
+        return [(point, extract_point_score(point)) for point in points]
 
     reranked_candidates = [
-        point
-        for _, point in sorted(
+        (point, float(score))
+        for score, point in sorted(
             zip(scores, rerank_candidates),
             key=lambda item: item[0],
             reverse=True,
         )
     ]
+    tail_with_scores = [
+        (point, extract_point_score(point))
+        for point in tail_candidates
+    ]
 
-    return reranked_candidates + retrieval_tail
+    return reranked_candidates + tail_with_scores
 
 
-def aggregate_message_ids(points: list[Any], top_k: int = FINAL_TOP_K) -> list[str]:
-    best_scores: dict[str, float] = {}
+def aggregate_message_scores(
+    scored_points: list[tuple[Any, float]],
+) -> tuple[dict[str, float], dict[str, int]]:
+    aggregated_scores: dict[str, float] = {}
     first_seen_rank: dict[str, int] = {}
-    total_points = len(points)
 
-    for rank, point in enumerate(points):
-        retrieval_score = float(getattr(point, "score", 0.0) or 0.0)
-        rank_score = float(total_points - rank)
-        score = rank_score + retrieval_score
+    for rank, (point, point_score) in enumerate(scored_points):
+        point_message_ids = set(extract_message_ids(point))
+        for message_id in point_message_ids:
+            aggregated_scores[message_id] = aggregated_scores.get(message_id, 0.0) + point_score
+            first_seen_rank.setdefault(message_id, rank)
 
-        for message_id in extract_message_ids(point):
-            if message_id not in first_seen_rank:
-                first_seen_rank[message_id] = rank
+    return aggregated_scores, first_seen_rank
 
-            if score > best_scores.get(message_id, float("-inf")):
-                best_scores[message_id] = score
 
-    ranked_ids = sorted(
-        best_scores,
-        key=lambda message_id: (-best_scores[message_id], first_seen_rank[message_id]),
+def select_top_message_ids(
+    aggregated_scores: dict[str, float],
+    first_seen_rank: dict[str, int],
+    limit: int,
+) -> list[str]:
+    sorted_items = sorted(
+        aggregated_scores.items(),
+        key=lambda item: (-item[1], first_seen_rank[item[0]]),
     )
-    return ranked_ids[:top_k]
+    return [message_id for message_id, _ in sorted_items[:limit]]
 
 
 # Ваш сервис должен имплементировать оба этих метода
@@ -445,11 +480,12 @@ async def search(payload: SearchAPIRequest) -> SearchAPIResponse:
     sparse_vectors = [await embed_sparse(item) for item in sparse_queries]
     best_points = await qdrant_search(qdrant, dense_vectors, sparse_vectors, payload.question)
 
-    if best_points is None:
+    if not best_points:
         return SearchAPIResponse(results=[])
 
-    best_points = await rerank_points(client, query, list(best_points))
-    message_ids = aggregate_message_ids(best_points)
+    scored_points = await rerank_points(client, query, deduplicate_points(list(best_points)))
+    aggregated_scores, first_seen_rank = aggregate_message_scores(scored_points)
+    message_ids = select_top_message_ids(aggregated_scores, first_seen_rank, FINAL_TOP_K)
 
     return SearchAPIResponse(
         results=[SearchAPIItem(message_ids=message_ids)]