main #5

Closed
jeffrey228 wants to merge 3 commits from jeffrey228/vk_hackathon:main into main
2 changed files with 96 additions and 39 deletions
Showing only changes of commit 408c7df005 - Show all commits

View file

@ -23,13 +23,37 @@
## Договоренности ## Договоренности
- Не менять контракты `POST /index`, `POST /sparse_embedding`, `POST /search`. - Не менять контракты `POST /index`, `POST /sparse_embedding`, `POST /search`.
- По текущей задаче фокус держать на `index`. - По текущей задаче фокус держать на `index`, если пользователь не просит иначе.
- Не трогать `search`, `docker-compose.yml` и инфраструктуру без отдельной просьбы. - Не трогать `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 запросы. - `question.variants`, `question.hyde`, `question.keywords` и entities используются как дополнительные dense/sparse запросы.
- Retrieval делает несколько prefetch-запросов и fusion через Qdrant. - Retrieval делает несколько prefetch-запросов и fusion через Qdrant.
- Rerank больше не выбрасывает retrieval-хвост. - Rerank больше не выбрасывает retrieval-хвост.
- Финальные `message_ids` агрегируются, дедуплицируются и ограничиваются `top-50`. - Retrieval points дедуплицируются по Qdrant point id.
- Финальные `message_ids` агрегируются по score, дедуплицируются и ограничиваются `top-50`.
Измененные файлы: Измененные файлы:
@ -86,10 +111,10 @@
Контекст по текущему состоянию: Контекст по текущему состоянию:
- `index/main.py` сейчас использует символьный chunking. - `index/main.py` тогда использовал символьный chunking.
- `render_message` берет только `message.text` и `parts[*].text`. - `render_message` брал только `message.text` и `parts[*].text`.
- `page_content`, `dense_content`, `sparse_content` сейчас одинаковые. - `page_content`, `dense_content`, `sparse_content` были одинаковые.
- В примере `data/Go Nova.json` текущий `build_chunks` покрывает 24 из 25 сообщений; системное сообщение с `member_event` выпадает из индекса. - В примере `data/Go Nova.json` старый `build_chunks` покрывал 24 из 25 сообщений; системное сообщение с `member_event` выпадало из индекса.
Измененные файлы: Измененные файлы:
@ -98,7 +123,3 @@
Проверки: Проверки:
- Код не запускался, потому что создан только журнал. - Код не запускался, потому что создан только журнал.
Следующий ожидаемый фокус:
- Перестроить индексный renderer и chunking в `index/main.py`, если пользователь попросит перейти к реализации.

View file

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